<!DOCTYPE html>


<html lang="zh_CN">
  

    <head>
      <meta charset="utf-8" />
        
      <meta
        name="viewport"
        content="width=device-width, initial-scale=1, maximum-scale=1"
      />
      <title>ActiveMQ笔记 |  MyBlog</title>
  <meta name="generator" content="hexo-theme-ayer">
      
      <link rel="shortcut icon" href="/favicon.ico" />
       
<link rel="stylesheet" href="/dist/main.css">

      <link
        rel="stylesheet"
        href="https://cdn.jsdelivr.net/gh/Shen-Yu/cdn/css/remixicon.min.css"
      />
      
<link rel="stylesheet" href="/css/custom.css">
 
      <script src="https://cdn.jsdelivr.net/npm/pace-js@1.0.2/pace.min.js"></script>
       
 

      <link
        rel="stylesheet"
        href="https://cdn.jsdelivr.net/npm/@sweetalert2/theme-bulma@5.0.1/bulma.min.css"
      />
      <script src="https://cdn.jsdelivr.net/npm/sweetalert2@11.0.19/dist/sweetalert2.min.js"></script>

      <!-- mermaid -->
      
      <style>
        .swal2-styled.swal2-confirm {
          font-size: 1.6rem;
        }
      </style>
    <link rel="alternate" href="/atom.xml" title="MyBlog" type="application/atom+xml">
</head>
  </html>
</html>


<body>
  <div id="app">
    
      
    <main class="content on">
      <section class="outer">
  <article
  id="post-ActiveMQ笔记"
  class="article article-type-post"
  itemscope
  itemprop="blogPost"
  data-scroll-reveal
>
  <div class="article-inner">
    
    <header class="article-header">
       
<h1 class="article-title sea-center" style="border-left:0" itemprop="name">
  ActiveMQ笔记
</h1>
 

      
    </header>
     
    <div class="article-meta">
      <a href="/2021/12/16/ActiveMQ%E7%AC%94%E8%AE%B0/" class="article-date">
  <time datetime="2021-12-16T02:37:53.000Z" itemprop="datePublished">2021-12-16</time>
</a>   
<div class="word_count">
    <span class="post-time">
        <span class="post-meta-item-icon">
            <i class="ri-quill-pen-line"></i>
            <span class="post-meta-item-text"> 字数统计:</span>
            <span class="post-count">11.8k</span>
        </span>
    </span>

    <span class="post-time">
        &nbsp; | &nbsp;
        <span class="post-meta-item-icon">
            <i class="ri-book-open-line"></i>
            <span class="post-meta-item-text"> 阅读时长≈</span>
            <span class="post-count">51 分钟</span>
        </span>
    </span>
</div>
 
    </div>
      
    <div class="tocbot"></div>




  
    <div class="article-entry" itemprop="articleBody">
       
  <h1 id="一、MQ的相关概念"><a href="#一、MQ的相关概念" class="headerlink" title="一、MQ的相关概念"></a>一、MQ的相关概念</h1><h2 id="1-1-1-什么是MQ"><a href="#1-1-1-什么是MQ" class="headerlink" title="1.1.1. 什么是MQ"></a>1.1.1. 什么是MQ</h2><p>MQ(message queue)，从字面意思上看，本质是个队列，FIFO先入先出，只不过队列中存放的内容是message而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了MQ之后，消息发送上游只需要依赖MQ，不用依赖其他服务。 </p>
<h2 id="1-1-2-为什么要用MQ"><a href="#1-1-2-为什么要用MQ" class="headerlink" title="1.1.2.为什么要用MQ"></a>1.1.2.为什么要用MQ</h2><h3 id="1-流量消峰"><a href="#1-流量消峰" class="headerlink" title="1.流量消峰"></a>1.流量消峰</h3><p>使访问的人进行排队，防止订单系统宕机</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112161103427.png" alt="image-20211216110323639"></p>
<h3 id="2-应用解耦"><a href="#2-应用解耦" class="headerlink" title="2.应用解耦"></a>2.应用解耦</h3><p>以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障，提示系统的可用性。</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112161316052.png" alt="image-20211216131630841"></p>
<h3 id="3-异步处理"><a href="#3-异步处理" class="headerlink" title="3.异步处理"></a>3.异步处理</h3><p>A调用B服务后，只需要监听B处理完成的消息，当B处理完成后，会发送一条消息给MQ，MQ会将此消息转发给A服务。这样A服务既不用循环调用B的查询API，也不要</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112161900728.png" alt="image-20211216190058422"></p>
<h2 id="1-1-3-MQ的分类"><a href="#1-1-3-MQ的分类" class="headerlink" title="1.1.3.MQ的分类"></a>1.1.3.MQ的分类</h2><h3 id="1-ActiveMQ"><a href="#1-ActiveMQ" class="headerlink" title="1.ActiveMQ"></a>1.ActiveMQ</h3><p>优点：单机吞吐量万级，时效性ms级，可用性高，基于主从架构实现高可用性。</p>
<p>缺点:官方社区现在对ActiveMQ 5.x维护越来越少，高吞吐量场景较少使用。 </p>
<h3 id="2-Kafka"><a href="#2-Kafka" class="headerlink" title="2.Kafka"></a>2.Kafka</h3><p>大数据的杀手锏，谈到大数据领域内的消息传输，则绕不开Kafka，这款为大数据而生的消息中间件，以其百万级TPS的吞吐量名声大噪，迅速成为大数据领域的宠儿，在数据采集、传输、存储的过程中发挥着举足轻重的作用。目前已经被LinkedIn，Uber, Twitter, Netflix等大公司所采纳。  </p>
<p>优点: 性能卓越，单机写入TPS约在百万条/秒，最大的优点，就是吞吐量高。时效性ms级可用性非常高，kafka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用,消费者采用Pull方式获取消息, 消息有序, 通过控制能够保证所有消息被消费且仅被消费一次;有优秀的第三方Kafka Web管理界面Kafka-Manager；在日志领域比较成熟，被多家公司和多个开源项目使用；功能支持：功能较为简单，主要支持简单的MQ功能，在大数据领域的实时计算以及日志采集被大规模使用  </p>
<p>缺点：Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消息响应时间变长，使用短轮询方式，实时性取决于轮询间隔时间，消费失败不支持重试；支持消息顺序，但是一台代理宕机后，就会产生消息乱序，社区更新较慢； </p>
<h3 id="3-RocketMQ"><a href="#3-RocketMQ" class="headerlink" title="3.RocketMQ"></a>3.RocketMQ</h3><p>RocketMQ出自阿里巴巴的开源产品，用Java语言实现，在设计时参考了 Kafka，并做出了自己的一些改进。被阿里巴巴广泛应用在订单，交易，充值，流计算，消息推送，日志流式处理，binglog分发等场景。</p>
<p> 优点:单机吞吐量十万级,可用性非常高，分布式架构,消息可以做到0丢失,MQ功能较为完善，还是分布式的，扩展性好,支持10亿级别的消息堆积，不会因为堆积导致性能下降,源码是java我们可以自己阅读源码，定制自己公司的MQ  </p>
<p>缺点：支持的客户端语言不多，目前是java及c++，其中c++不成熟；社区活跃度一般,没有在MQ核心中去实现JMS等接口,有些系统要迁移需要修改大量代码 </p>
<h3 id="4-RabbitMQ"><a href="#4-RabbitMQ" class="headerlink" title="4.RabbitMQ"></a>4.RabbitMQ</h3><p>2007年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。 </p>
<p> 优点:由于erlang语言的高并发特性，性能较好；吞吐量到万级，MQ功能比较完备,健壮、稳定、易用、跨平台、支持多种语言 如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX文档齐全；开源提供的管理界面非常棒，用起来很好用,社区活跃度高；更新频率相当高 <a target="_blank" rel="noopener" href="https://www.rabbitmq.com/news.html">https://www.rabbitmq.com/news.html</a>  </p>
<p>缺点：商业版需要收费,学习成本较高 </p>
<h1 id="二、面试题"><a href="#二、面试题" class="headerlink" title="二、面试题"></a>二、面试题</h1><blockquote>
<p>在何种场景下使用消息中间件</p>
</blockquote>
<p>流量高峰期</p>
<blockquote>
<p>为什么要在系统里引入消息中间件</p>
</blockquote>
<ul>
<li>解决耦合调用的问题</li>
<li>流量削峰，达到保护主业务的目的</li>
<li>异步</li>
</ul>
<p>消息队列的大致过程：</p>
<p>发送者把消息发送给消息服务器，消息服务器将消息存放在若干<font color='red'>队列/主题</font>中，在合适的时候，消息服务器会将消息转发给接受者。在这个过程中，发送和接受是异步的，也就是发送无需等待，而且发送者和接受者的生命周期没有必然关系</p>
<h1 id="三、ActiveMQ最主要的功能"><a href="#三、ActiveMQ最主要的功能" class="headerlink" title="三、ActiveMQ最主要的功能"></a>三、ActiveMQ最主要的功能</h1><p>实现高可用、高性能、可伸缩、易用和安全的企业级面向消息服务的系统</p>
<h2 id="ActiveMQ的技术维度"><a href="#ActiveMQ的技术维度" class="headerlink" title="ActiveMQ的技术维度"></a>ActiveMQ的技术维度</h2><ul>
<li>api发送和接收</li>
<li>MQ的高可用性</li>
<li>MQ的集群和容错配置</li>
<li>MQ的持久化</li>
<li>延时发送/定时投递</li>
<li>签收机制</li>
<li>Spring整合</li>
<li>编程语言：java</li>
</ul>
<h1 id="四、安装"><a href="#四、安装" class="headerlink" title="四、安装"></a>四、安装</h1><p>1.官网下载</p>
<p><a target="_blank" rel="noopener" href="https://activemq.apache.org/components/classic/download/">https://activemq.apache.org/components/classic/download/</a></p>
<p>2.安装到linux的 /opt 目录下</p>
<p>3.解压缩tar -zxvf apache-activemq-5.16.3-bin.tar.gz</p>
<p>4.cp -r /opt/apache-activemq-5.16.3 /myactiveMQ/</p>
<p>5.启动activeMQ</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112162057401.png" alt="image-20211216205750770"></p>
<p>6.activeMQ的默认进程端口61616</p>
<p>7.查看后台进程  lsof -i:61616</p>
<p>后台进程61616</p>
<p>前台访问地址<a target="_blank" rel="noopener" href="http://192.168.227.20:8161/admin%EF%BC%88%E9%9C%80%E8%A6%81%E5%85%B3%E9%97%AD%E9%98%B2%E7%81%AB%E5%9C%88%EF%BC%8C%E6%88%96%E8%80%85%E5%BC%80%E6%94%BE%E7%AB%AF%E5%8F%A3%EF%BC%89">http://192.168.227.20:8161/admin（需要关闭防火圈，或者开放端口）</a></p>
<p><strong>备注：1.采用61616端口提供JMS服务</strong></p>
<p><strong>2.采用8161提供管理控制台服务</strong></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112162136519.png" alt="image-20211216213648842"></p>
<p><font color='red'>用户名和密码都为:admin</font></p>
<h1 id="五、Java编码实现ActiveMQ通讯"><a href="#五、Java编码实现ActiveMQ通讯" class="headerlink" title="五、Java编码实现ActiveMQ通讯"></a>五、Java编码实现ActiveMQ通讯</h1><p>IDEA建Maven工程</p>
<p>导入依赖</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependencies</span>&gt;</span></span><br><span class="line">      <span class="comment">&lt;!--activemq所需要的的jar包配置--&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.activemq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>activemq-all<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.15.9<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.xbean<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>xbean-spring<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>3.16<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="comment">&lt;!--下面是junit/log4j等基础通用配置--&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.slf4j<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>slf4j-api<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.7.6<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>ch.qos.logback<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>logback-classic<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.2.3<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.projectlombok<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>lombok<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.16.18<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">scope</span>&gt;</span>provided<span class="tag">&lt;/<span class="name">scope</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>junit<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>junit<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>4.11<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">scope</span>&gt;</span>test<span class="tag">&lt;/<span class="name">scope</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">  <span class="tag">&lt;/<span class="name">dependencies</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>JMS编码总体架构</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112170920661.png" alt="image-20211217092035417"></p>
<ul>
<li>在点对点的消息传递域中，目的地被称为队列（queue）</li>
</ul>
<p>点对点消息传递域的特点如下：</p>
<p>1）每个消息只能有一个消费者，类似1对1的关系。</p>
<p>2）消息的生产者和消费者之间<font color='red'>没有时间上的相关性</font>。无论消费者在生产者发送消息的时候是否处在运行状态，消费者都可以提取消息。</p>
<p>3）消息被消费后队列中<font color='red'>不会再存储</font>，所以消费者<font color='red'>不会消费到已经被消费掉的消息</font></p>
<ul>
<li>在发布订阅消息传递域中，目的地被称为主题（topic）</li>
</ul>
<p>发布订阅传递域的特点如下：</p>
<p>1）生产者将消息发布到topic中，每个消息可以有多个消费者，属于一对多</p>
<p>2）生产者和消费者之间有时间上的相关性。订阅某个主题的消费者只能消费自它订阅之后发布的消息。</p>
<p>3）<font color='red'>生产者生产时，topic不保存消息</font>它是无状态的不落地，假如无人订阅就去生产，那就是一条废消息，所以，一般先启动消费者再启动生产者</p>
<h2 id="1-点对点消息传递"><a href="#1-点对点消息传递" class="headerlink" title="1.点对点消息传递"></a>1.点对点消息传递</h2><blockquote>
<p>消息生产者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsProduce</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue01&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//5.创建消息的生产者</span></span><br><span class="line">        MessageProducer messageProducer = session.createProducer(queue);</span><br><span class="line">        <span class="comment">//6.通过使用MEssageProducer生产三条消息发送到MQ的队列里面</span></span><br><span class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">1</span>; i &lt;= <span class="number">3</span>; i++) &#123;</span><br><span class="line">            <span class="comment">//7.创建消息</span></span><br><span class="line">            TextMessage textMessage = session.createTextMessage(<span class="string">&quot;msg--&quot;</span> + i);</span><br><span class="line">            <span class="comment">//8.通过messageProducer发送给mq</span></span><br><span class="line">            messageProducer.send(textMessage);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//9.关闭资源</span></span><br><span class="line">        messageProducer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">        System.out.println(<span class="string">&quot;=====消息发布到MQ完成=====&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112171016891.png" alt="image-20211217101642229"></p>
<table>
<thead>
<tr>
<th>Number Of Pending</th>
<th>Number Of Consumers</th>
<th>Messages Enqueued</th>
<th>Messages Dequeues</th>
</tr>
</thead>
<tbody><tr>
<td>等待消费的消息</td>
<td>消费者数量</td>
<td>进队消息数</td>
<td>出队消息数</td>
</tr>
<tr>
<td>这个是当前未出队列的数量。公式=总接收数 - 总出队列数</td>
<td>消费者端的消费者数量</td>
<td>进入队列的总数量，包括出队列的。这个数量只增不减</td>
<td>可以理解为是消费者掉的数量</td>
</tr>
</tbody></table>
<p>总结：</p>
<p>当有一个消息进入这个队列时，等待消费的消息是1，进入队列的消息是1.</p>
<p>当消息消费后，等待消费的消息是0，进入队列的消息是1，出队列的消息是1</p>
<p>再来一条消息时，等待消费的消息是1，进入队列的消息是2</p>
<blockquote>
<p>消息消费者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsConsumer</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue01&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//创建消费者</span></span><br><span class="line">        MessageConsumer consumer = session.createConsumer(queue);</span><br><span class="line">        <span class="keyword">while</span>(<span class="keyword">true</span>)&#123;</span><br><span class="line">            TextMessage textMessage = (TextMessage) consumer.receive();</span><br><span class="line">            <span class="keyword">if</span>(textMessage != <span class="keyword">null</span>)&#123;</span><br><span class="line">                System.out.println(<span class="string">&quot;=====消费者接收到消息:&quot;</span>+textMessage.getText()+<span class="string">&quot;=====&quot;</span>);</span><br><span class="line">            &#125;<span class="keyword">else</span>&#123;</span><br><span class="line">                <span class="keyword">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        consumer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112171130658.png" alt="image-20211217113030076"></p>
<p><strong>同步阻塞方式（receive方法）</strong></p>
<blockquote>
<p>订阅者或接收者调用MessageConsumer的receive()方法来接收消息，receive方法在能够接收到消息之前（或超时之前）将一直阻塞。</p>
</blockquote>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112171135219.png" alt="image-20211217113503800"></p>
<ul>
<li>无参方法表示一直等待</li>
<li>带参方法表示 xx毫秒后不再等待</li>
</ul>
<blockquote>
<p>通过监听的方式来消费消息 </p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line">consumer.setMessageListener(<span class="keyword">new</span> MessageListener() &#123;</span><br><span class="line">           <span class="meta">@Override</span></span><br><span class="line">           <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onMessage</span><span class="params">(Message message)</span> </span>&#123;</span><br><span class="line">               <span class="keyword">if</span>(message != <span class="keyword">null</span> &amp;&amp; message <span class="keyword">instanceof</span> TextMessage)&#123;</span><br><span class="line">                   TextMessage textMessage = (TextMessage) message;</span><br><span class="line">                   <span class="keyword">try</span> &#123;</span><br><span class="line">                       System.out.println(<span class="string">&quot;=====消费者接收到消息:&quot;</span>+textMessage.getText()+<span class="string">&quot;=====&quot;</span>);</span><br><span class="line">                   &#125; <span class="keyword">catch</span> (JMSException e) &#123;</span><br><span class="line">                       System.err.println(e.getMessage());</span><br><span class="line">                   &#125;</span><br><span class="line">               &#125;</span><br><span class="line">           &#125;</span><br><span class="line">       &#125;);</span><br><span class="line">       <span class="comment">//阻塞</span></span><br><span class="line">       System.in.read();</span><br><span class="line">       consumer.close();</span><br><span class="line">       session.close();</span><br><span class="line">       connection.close();</span><br></pre></td></tr></table></figure>

<h3 id="消费者三大消费情况"><a href="#消费者三大消费情况" class="headerlink" title="消费者三大消费情况"></a>消费者三大消费情况</h3><ul>
<li>先生产，再开启1号消费者：1号消费者能接收到消息</li>
<li>先生产，然后开启1号消费者再开启2号消费者：2号消费者不能接收到消息</li>
<li>先开启1号消费者，然后开启2号消费者，再生产：最后会以轮询的方式接收消息</li>
</ul>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112181413313.png" alt="image-20211218141329259"></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112181413527.png" alt="image-20211218141353410"></p>
<h2 id="2-发布订阅传递（Topic）"><a href="#2-发布订阅传递（Topic）" class="headerlink" title="2.发布订阅传递（Topic）"></a>2.发布订阅传递（Topic）</h2><blockquote>
<p>生产者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">在点对点的代码基础上进行修改即可，需要修改的内容如下：</span><br><span class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TOPIC_NAME = <span class="string">&quot;topic-yuan&quot;</span>;</span><br><span class="line"></span><br><span class="line"><span class="comment">//4.创建目的地</span></span><br><span class="line">Topic topic = session.createTopic(TOPIC_NAME);</span><br></pre></td></tr></table></figure>

<blockquote>
<p>消费者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">需要修改的内容与上面相同...</span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112181454539.png" alt="image-20211218145435014"></p>
<p><font color='red'><strong>如果先生产，再订阅，将无法消费之前的消息。</strong></font></p>
<h1 id="六、两大模式比较（面试题）"><a href="#六、两大模式比较（面试题）" class="headerlink" title="六、两大模式比较（面试题）"></a>六、两大模式比较（面试题）</h1><table>
<thead>
<tr>
<th>比较项目</th>
<th>Topic 模式队列</th>
<th>Queue 模式队列</th>
</tr>
</thead>
<tbody><tr>
<td>工作模式</td>
<td>“订阅-发布”模式，如果当前没有订阅者，消息将会被丢弃。如果有多个订阅者，那么这些订阅者都会收到消息</td>
<td>“负载均衡”模式，如果当前没有消费者，消息也不会丢弃；如果有多个消费者，那么一条消息也只会发送给期中一个消费者，并且要求消费者回复ack信息</td>
</tr>
<tr>
<td>有无状态</td>
<td>无状态</td>
<td>Queue数据默认会在mq服务器上以文件形式保存，比如ActiveMQ一般保存在$AMQ_HOME\data\kr-store\data下面，也可以配置成DB存储</td>
</tr>
<tr>
<td>传递完整性</td>
<td>如果没有订阅者，消息会被丢弃</td>
<td>消息不会被丢弃</td>
</tr>
<tr>
<td>处理效率</td>
<td>由于消息要按照订阅者的数量进行赋值，所以处理性能会随着订阅者的增加而明显降低，并且还要结合不同消息协议自身的新能差异</td>
<td>由于一条消息只发给一个消费者，所以就算消费者再多，性能也不会有明显降低。当然不同消息协议的具体性能也是有差异的</td>
</tr>
</tbody></table>
<h1 id="七、JMS规范和落地产品"><a href="#七、JMS规范和落地产品" class="headerlink" title="七、JMS规范和落地产品"></a>七、JMS规范和落地产品</h1><h2 id="1-JMS定义"><a href="#1-JMS定义" class="headerlink" title="1.JMS定义"></a>1.JMS定义</h2><p>Java Message Service（Java消息服务是JavaEE的子规范 ）</p>
<h2 id="2-四大消息中间件的比较"><a href="#2-四大消息中间件的比较" class="headerlink" title="2.四大消息中间件的比较"></a>2.四大消息中间件的比较</h2><table>
<thead>
<tr>
<th>特性</th>
<th>ActiveMQ</th>
<th>RabbitMQ</th>
<th>Kafka</th>
<th>RocketMQ</th>
</tr>
</thead>
<tbody><tr>
<td>PRODUCER-COMSUMER</td>
<td>支持</td>
<td>支持</td>
<td>支持</td>
<td>支持</td>
</tr>
<tr>
<td>PUBLISH-SUBSCRIBE</td>
<td>支持</td>
<td>支持</td>
<td>支持</td>
<td>支持</td>
</tr>
<tr>
<td>REQUEST-REPLY</td>
<td>支持</td>
<td>支持</td>
<td>-</td>
<td>支持</td>
</tr>
<tr>
<td>API完备性</td>
<td>高</td>
<td>高</td>
<td>高</td>
<td>低（静态配置）</td>
</tr>
<tr>
<td>多语言支持</td>
<td>支持，JAVA优先</td>
<td>语言无关</td>
<td>支持，JAVA优先</td>
<td>支持</td>
</tr>
<tr>
<td>单机呑吐量</td>
<td>万级</td>
<td>万级</td>
<td>十万级</td>
<td>单机万级</td>
</tr>
<tr>
<td>消息延迟</td>
<td>-</td>
<td>微秒级</td>
<td>毫秒级</td>
<td>-</td>
</tr>
<tr>
<td>可用性</td>
<td>高（主从）</td>
<td>高（主从）</td>
<td>非常高（分布式）</td>
<td>高</td>
</tr>
<tr>
<td>消息丢失</td>
<td>-</td>
<td>低</td>
<td>理论上不会丢失</td>
<td>-</td>
</tr>
<tr>
<td>消息重复</td>
<td>-</td>
<td>可控制</td>
<td>理论上会有重复</td>
<td>-</td>
</tr>
<tr>
<td>文档的完备性</td>
<td>高</td>
<td>高</td>
<td>高</td>
<td>中</td>
</tr>
<tr>
<td>提供快速入门</td>
<td>有</td>
<td>有</td>
<td>有</td>
<td>无</td>
</tr>
<tr>
<td>首次部署难度</td>
<td>-</td>
<td>低</td>
<td>中</td>
<td>高</td>
</tr>
</tbody></table>
<h2 id="3-JMS组成的四大元素"><a href="#3-JMS组成的四大元素" class="headerlink" title="3.JMS组成的四大元素"></a>3.JMS组成的四大元素</h2><ul>
<li>JMS provider：消息中间件</li>
<li>JMS producer：消息生产者</li>
<li>JMS consumer：消息消费者</li>
<li>JMS message<ul>
<li>消息头 — extMessage.setXX()<ul>
<li>JMSDestination：消息发送的目的地，主要是指Queue和Topic</li>
<li>JMSDeliveryMode：持久和非持久模式</li>
<li>JMSExpiration：设置消息过期的时间，<font color='red'>默认永不过期</font>（前提：没有签收）</li>
<li>JMSPriority：消息优先级，从0-9十个级别，0到4是普通消息，5到9是加急消息。JMS不要求MQ严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。默认是4级。</li>
<li>JMSMessageID：唯一识别每个消息的标识，由消息中间件产生</li>
</ul>
</li>
<li>消息体 — session.createXX()<ul>
<li>封装具体的消息数据</li>
<li>5种消息体格式<ul>
<li><font color='red'>TextMessage</font>：普通字符串消息，包含一个string</li>
<li><font color='red'>MapMessage</font>：一个Map类型的消息，key为String类型，而值为java的基本类型</li>
<li>BytesMessage：二进制数组消息，包含一个byte[]</li>
<li>StreamMessage：Java数据流消息，用标准流操作来顺序的填充和读取</li>
<li>ObjectMessage：对象消息，包含一个可序列化的Java对象</li>
</ul>
</li>
<li>发送和接受的消息体类型必须一致对应</li>
</ul>
</li>
<li>消息属性—textMessage.setXX()<ul>
<li>如果需要除消息头字段以外的值，那么可以使用消息属性</li>
<li>识别/去重/重点标注等操作非常有用的方法</li>
<li>以属性名和属性值对的方式的形式制定</li>
</ul>
</li>
</ul>
</li>
</ul>
<h2 id="4、JMS的可靠性"><a href="#4、JMS的可靠性" class="headerlink" title="4、JMS的可靠性"></a>4、JMS的可靠性</h2><h3 id="1-PERSISTENT：持久性"><a href="#1-PERSISTENT：持久性" class="headerlink" title="1.PERSISTENT：持久性"></a>1.PERSISTENT：持久性</h3><h4 id="1-1-参数设置说明"><a href="#1-1-参数设置说明" class="headerlink" title="1.1 参数设置说明"></a>1.1 参数设置说明</h4><p>非持久：</p>
<ul>
<li>messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT)</li>
<li>非持久化：当服务器宕机，消息不存在</li>
</ul>
<p>持久：</p>
<ul>
<li>messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)</li>
<li>持久化：当服务器宕机，消息依然存在，但<font color='red'>消息入队的记录会被清除</font></li>
</ul>
<h4 id="1-2持久的Queue"><a href="#1-2持久的Queue" class="headerlink" title="1.2持久的Queue"></a>1.2持久的Queue</h4><p><strong><font color='red'>队列的默认传送模式是持久化</font></strong></p>
<h4 id="1-3持久的Topic"><a href="#1-3持久的Topic" class="headerlink" title="1.3持久的Topic"></a>1.3持久的Topic</h4><p>客户端首先向MQ注册一个自己的身份ID识别号，当这个客户端处于离线时，生产者会为这个ID保存所有发送到主题的消息，当客户再次连接到MQ时会根据消费者的ID得到所有当自己处于离线时发送到主题的消息</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br></pre></td><td class="code"><pre><span class="line">=====Produce_Topic调改位置=====</span><br><span class="line"></span><br><span class="line">messageProducer.setDeliveryMode();</span><br><span class="line">将connection.start();  </span><br><span class="line"></span><br><span class="line">=====Consumer_Topic调改=====</span><br><span class="line"></span><br><span class="line">connection.setClientID(<span class="string">&quot;XX&quot;</span>);</span><br><span class="line"><span class="comment">//订阅主题</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsConsumer_Topic_Persist</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String TOPIC_NAME = <span class="string">&quot;topic-yuan&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.setClientID(<span class="string">&quot;z3&quot;</span>);</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Topic topic = session.createTopic(TOPIC_NAME);</span><br><span class="line">        TopicSubscriber topicSubscription = session.createDurableSubscriber(topic,<span class="string">&quot;remark...&quot;</span>);</span><br><span class="line">        connection.start();</span><br><span class="line"></span><br><span class="line">        Message message = topicSubscription.receive();</span><br><span class="line">        <span class="keyword">while</span>(message != <span class="keyword">null</span>)&#123;</span><br><span class="line">            TextMessage textMessage = (TextMessage) message;</span><br><span class="line">            System.out.println(<span class="string">&quot;=====收到的持久化topic:&quot;</span>+textMessage.getText());</span><br><span class="line">            message = topicSubscription.receive(<span class="number">1000L</span>);</span><br><span class="line">        &#125;</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>一定要先运行一次消费者，等于向MQ注册，类似订阅了主题，然后再运行生产者发送信息，此时无论消费者是否在线，都会接收到，不在先的话，下次连接的时候，会把没有收过的消息都接收下来</p>
<h3 id="2-transaction：事务"><a href="#2-transaction：事务" class="headerlink" title="2.transaction：事务"></a>2.transaction：事务</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">Session session = connection.createSession(<span class="keyword">true</span>,Session.AUTO_ACKNOWLEDGE); <span class="comment">//开启事务</span></span><br><span class="line"><span class="comment">//需要session提交，消息才会入队</span></span><br><span class="line">session.commit();</span><br><span class="line"></span><br><span class="line">session.rollback();  <span class="comment">//回滚</span></span><br></pre></td></tr></table></figure>

<h3 id="3-acknowledge：签收"><a href="#3-acknowledge：签收" class="headerlink" title="3.acknowledge：签收"></a>3.acknowledge：签收</h3><h4 id="3-1非事务"><a href="#3-1非事务" class="headerlink" title="3.1非事务"></a>3.1非事务</h4><blockquote>
<p>自动签收（默认）</p>
</blockquote>
<p>Session.AUTO_ACKNOWLEDGE</p>
<blockquote>
<p>手动签收</p>
</blockquote>
<ul>
<li>Session.CLIENT_ACKNOWLEDGE</li>
<li>客户端调用acknowledge方法手动签收（  textMessage.acknowledge()  )</li>
</ul>
<blockquote>
<p>允许重复消息</p>
</blockquote>
<p>Session.DUPS_OK_ACKNOWLEDGE</p>
<h4 id="3-2事务"><a href="#3-2事务" class="headerlink" title="3.2事务"></a>3.2事务</h4><p>只要开启事务就需要提交，如果没有提交，就算ack签收了也会导致消息重复使用，没有被消费</p>
<h4 id="3-3签收和事务的关系"><a href="#3-3签收和事务的关系" class="headerlink" title="3.3签收和事务的关系"></a>3.3签收和事务的关系</h4><ol>
<li>在事务性会话种，当一个事务被成功提交则消息被自动签收。如果事务回滚，则消息会被再次传送</li>
<li>非事务性会话中，消息何时被确认取决于创建会话时的应答模式</li>
</ol>
<h1 id="八、ActiveMQ的Broker"><a href="#八、ActiveMQ的Broker" class="headerlink" title="八、ActiveMQ的Broker"></a>八、ActiveMQ的Broker</h1><h2 id="1-Broker是什么"><a href="#1-Broker是什么" class="headerlink" title="1.Broker是什么"></a>1.Broker是什么</h2><p>相当于一个ActiveMQ服务器实例</p>
<p>Broker其实就是实现了用代码的形式启动ActiveMQ将MQ嵌入得到Java代码中，一边随时用随时启动，在用的时候再去启动这样就能节省资源，也保证了可靠性</p>
<h2 id="2-嵌入式Broker"><a href="#2-嵌入式Broker" class="headerlink" title="2.嵌入式Broker"></a>2.嵌入式Broker</h2><p>用ActiveMQ Broker作为独立的消息服务器来构建JAVA应用</p>
<p>ActiveMQ也支持在vm种通信基于嵌入式的broker，能够无缝的集成其他java应用</p>
<p>1.pom.xml导入依赖</p>
<p>否则会抛异常</p>
<p>java.lang.ClassNotFoundException:com.fasterxml.jackson.databind.ObjectMapper</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.fasterxml.jackson.core<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>jackson-databind<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.10.2<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>嵌入式Broker启动，此时MQ是在本机启动，所以bindAddress要写localhost</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">EmberBroker</span> </span>&#123;</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">        BrokerService brokerService = <span class="keyword">new</span> BrokerService();</span><br><span class="line">        brokerService.setUseJmx(<span class="keyword">true</span>);</span><br><span class="line">        brokerService.addConnector(<span class="string">&quot;tcp://localhost:61616&quot;</span>);</span><br><span class="line">        brokerService.start();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>生产者</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsProducer</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://localhost:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue01&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//5.创建消息的生产者</span></span><br><span class="line">        MessageProducer messageProducer = session.createProducer(queue);</span><br><span class="line">        <span class="comment">//6.通过使用MEssageProducer生产三条消息发送到MQ的队列里面</span></span><br><span class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">1</span>; i &lt;= <span class="number">3</span>; i++) &#123;</span><br><span class="line">            <span class="comment">//7.创建消息</span></span><br><span class="line">            TextMessage textMessage = session.createTextMessage(<span class="string">&quot;MessageListener---&quot;</span> + i);</span><br><span class="line">            <span class="comment">//8.通过messageProducer发送给mq</span></span><br><span class="line">            messageProducer.send(textMessage);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//9.关闭资源</span></span><br><span class="line">        messageProducer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">        System.out.println(<span class="string">&quot;=====消息发布到MQ完成=====&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>消费者</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsConsumer</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://localhost:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue01&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//创建消费者</span></span><br><span class="line">        MessageConsumer consumer = session.createConsumer(queue);</span><br><span class="line">        <span class="comment">/*while(true)&#123;</span></span><br><span class="line"><span class="comment">            TextMessage textMessage = (TextMessage) consumer.receive();</span></span><br><span class="line"><span class="comment">            if(textMessage != null)&#123;</span></span><br><span class="line"><span class="comment">                System.out.println(&quot;=====消费者接收到消息:&quot;+textMessage.getText()+&quot;=====&quot;);</span></span><br><span class="line"><span class="comment">            &#125;else&#123;</span></span><br><span class="line"><span class="comment">                break;</span></span><br><span class="line"><span class="comment">            &#125;</span></span><br><span class="line"><span class="comment">        &#125;</span></span><br><span class="line"><span class="comment">        */</span></span><br><span class="line">        System.out.println(<span class="string">&quot;2号消费者&quot;</span>);</span><br><span class="line">        consumer.setMessageListener(<span class="keyword">new</span> MessageListener() &#123;</span><br><span class="line">            <span class="meta">@Override</span></span><br><span class="line">            <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onMessage</span><span class="params">(Message message)</span> </span>&#123;</span><br><span class="line">                <span class="keyword">if</span>(message != <span class="keyword">null</span> &amp;&amp; message <span class="keyword">instanceof</span> TextMessage)&#123;</span><br><span class="line">                    TextMessage textMessage = (TextMessage) message;</span><br><span class="line">                    <span class="keyword">try</span> &#123;</span><br><span class="line">                        System.out.println(<span class="string">&quot;=====消费者接收到消息:&quot;</span>+textMessage.getText()+<span class="string">&quot;=====&quot;</span>);</span><br><span class="line">                    &#125; <span class="keyword">catch</span> (JMSException e) &#123;</span><br><span class="line">                        System.err.println(e.getMessage());</span><br><span class="line">                    &#125;</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;);</span><br><span class="line">        <span class="comment">//阻塞</span></span><br><span class="line">        System.in.read();</span><br><span class="line">        consumer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h1 id="九、Spring整合ActiveMQ"><a href="#九、Spring整合ActiveMQ" class="headerlink" title="九、Spring整合ActiveMQ"></a>九、Spring整合ActiveMQ</h1><h2 id="1-添加Spring支持JMS的包"><a href="#1-添加Spring支持JMS的包" class="headerlink" title="1.添加Spring支持JMS的包"></a>1.添加Spring支持JMS的包</h2><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>com.fasterxml.jackson.core<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>jackson-databind<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.10.2<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-jms<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.2.18.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.apache.activemq<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>activemq-pool<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.15.9<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line"><span class="comment">&lt;!--Spring-Aop相关jar包--&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-core<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.2.5.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-context<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.2.5.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-aop<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.2.5.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="comment">&lt;!-- https://mvnrepository.com/artifact/org.springframework/spring-orm --&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-orm<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>5.2.12.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line"></span><br><span class="line">      <span class="comment">&lt;!-- https://mvnrepository.com/artifact/org.aspectj/aspectjrt --&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.aspectj<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>aspectjrt<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.9.6<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line"></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>aspectj<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>aspectjweaver<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>1.5.3<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>cglib<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>cglib<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.1_2<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="2-Spring配置文件applicationContext-xml"><a href="#2-Spring配置文件applicationContext-xml" class="headerlink" title="2.Spring配置文件applicationContext.xml"></a>2.Spring配置文件applicationContext.xml</h2><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">&lt;?xml version=&quot;1.0&quot; encoding=&quot;UTF-8&quot;?&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">beans</span> <span class="attr">xmlns</span>=<span class="string">&quot;http://www.springframework.org/schema/beans&quot;</span></span></span><br><span class="line"><span class="tag">       <span class="attr">xmlns:xsi</span>=<span class="string">&quot;http://www.w3.org/2001/XMLSchema-instance&quot;</span></span></span><br><span class="line"><span class="tag">       <span class="attr">xmlns:context</span>=<span class="string">&quot;http://www.springframework.org/schema/context&quot;</span></span></span><br><span class="line"><span class="tag">       <span class="attr">xsi:schemaLocation</span>=<span class="string">&quot;http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd&quot;</span>&gt;</span></span><br><span class="line"></span><br><span class="line">    <span class="comment">&lt;!--开启包的自动扫描--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">context:component-scan</span> <span class="attr">base-package</span>=<span class="string">&quot;com.atyuan.activemq&quot;</span>/&gt;</span></span><br><span class="line">    <span class="comment">&lt;!--配置生产者--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;jmsFactory&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.activemq.pool.PooledConnectionFactory&quot;</span> <span class="attr">destroy-method</span>=<span class="string">&quot;stop&quot;</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;connectionFactory&quot;</span>&gt;</span></span><br><span class="line">            <span class="comment">&lt;!--真正可以产生Connection的ConnectionFactory，由对应的JMS服务厂商提供--&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">bean</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.activemq.ActiveMQConnectionFactory&quot;</span>&gt;</span></span><br><span class="line">                <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;brokerURL&quot;</span> <span class="attr">value</span>=<span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>/&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;maxConnections&quot;</span> <span class="attr">value</span>=<span class="string">&quot;100&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line"></span><br><span class="line">    <span class="comment">&lt;!--队列的目的地，点对点--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;destinationQueue&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.activemq.command.ActiveMQQueue&quot;</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">constructor-arg</span> <span class="attr">index</span>=<span class="string">&quot;0&quot;</span> <span class="attr">value</span>=<span class="string">&quot;spring-active-queue&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line">    </span><br><span class="line">    <span class="comment">&lt;!--主题，发布订阅多对一--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;destinationTopic&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.activemq.command.ActiveMQTopic&quot;</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">constructor-arg</span> <span class="attr">index</span>=<span class="string">&quot;0&quot;</span> <span class="attr">value</span>=<span class="string">&quot;spring-active-topic&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line"></span><br><span class="line">    <span class="comment">&lt;!--Spring提供的JMS工具类，它可以进行消息发送、接收等--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;jmsTemplate&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.springframework.jms.core.JmsTemplate&quot;</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;connectionFactory&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;jmsFactory&quot;</span>/&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;defaultDestination&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;destinationQueue&quot;</span>/&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;messageConverter&quot;</span>&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">bean</span> <span class="attr">class</span>=<span class="string">&quot;org.springframework.jms.support.converter.SimpleMessageConverter&quot;</span>/&gt;</span></span><br><span class="line">        <span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">beans</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="3-队列"><a href="#3-队列" class="headerlink" title="3.队列"></a>3.队列</h2><blockquote>
<p>生产者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SpringMQ_Produce</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> JmsTemplate jmsTemplate;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> </span>&#123;</span><br><span class="line">        ApplicationContext applicationContext = <span class="keyword">new</span> ClassPathXmlApplicationContext(<span class="string">&quot;applicationContext.xml&quot;</span>);</span><br><span class="line">        SpringMQ_Produce springMQ_produce = applicationContext.getBean(SpringMQ_Produce.class);</span><br><span class="line">        springMQ_produce.jmsTemplate.send(session -&gt; &#123;</span><br><span class="line">                TextMessage textMessage = session.createTextMessage(<span class="string">&quot;=====整合spring创建message=====&quot;</span>);</span><br><span class="line">                <span class="keyword">return</span> textMessage;</span><br><span class="line">            &#125;);</span><br><span class="line">        System.out.println(<span class="string">&quot;====send task over====&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<blockquote>
<p>消费者</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">SpringMQ_Consumer</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> JmsTemplate jmsTemplate;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> </span>&#123;</span><br><span class="line">        ApplicationContext applicationContext = <span class="keyword">new</span> ClassPathXmlApplicationContext(<span class="string">&quot;applicationContext.xml&quot;</span>);</span><br><span class="line">        SpringMQ_Consumer springMQ_consumer = applicationContext.getBean(SpringMQ_Consumer.class);</span><br><span class="line">        String retValue = (String) springMQ_consumer.jmsTemplate.receiveAndConvert();</span><br><span class="line"></span><br><span class="line">        System.out.println(<span class="string">&quot;消费者收到的消息:\n&quot;</span>+retValue);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h2 id="4-主题"><a href="#4-主题" class="headerlink" title="4.主题"></a>4.主题</h2><p>4.1修改配置文件</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line">===新增主题===</span><br><span class="line"><span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;destinationTopic&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.activemq.command.ActiveMQTopic&quot;</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;<span class="name">constructor-arg</span> <span class="attr">index</span>=<span class="string">&quot;0&quot;</span> <span class="attr">value</span>=<span class="string">&quot;spring-active-topic&quot;</span>/&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br><span class="line"></span><br><span class="line">=====修改=====</span><br><span class="line"> <span class="comment">&lt;!--Spring提供的JMS工具类，它可以进行消息发送、接收等--&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;jmsTemplate&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.springframework.jms.core.JmsTemplate&quot;</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;connectionFactory&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;jmsFactory&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;defaultDestination&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;destinationTopic&quot;</span>/&gt;</span>  <span class="comment">&lt;!--修改目的地--&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;messageConverter&quot;</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">bean</span> <span class="attr">class</span>=<span class="string">&quot;org.springframework.jms.support.converter.SimpleMessageConverter&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">property</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="5-监听器配置"><a href="#5-监听器配置" class="headerlink" title="5.监听器配置"></a>5.监听器配置</h2><p>在Spring里面实现消费者不启动，直接通过配置监听完成</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">&lt;!--配置监听程序--&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;jmsContainer&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.springframework.jms.listener.DefaultMessageListenerContainer&quot;</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;connectionFactory&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;jmsFactory&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;destination&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;destinationTopic&quot;</span>/&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;messageListener&quot;</span> <span class="attr">ref</span>=<span class="string">&quot;myMessageListener&quot;</span>/&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br></pre></td></tr></table></figure>

<blockquote>
<p>MessageListener实现类的编写</p>
</blockquote>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyMessageListener</span> <span class="keyword">implements</span> <span class="title">MessageListener</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Override</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onMessage</span><span class="params">(Message message)</span> </span>&#123;</span><br><span class="line">        <span class="keyword">if</span>(<span class="keyword">null</span> != message &amp;&amp; message <span class="keyword">instanceof</span> TextMessage)&#123;</span><br><span class="line">            TextMessage textMessage = (TextMessage) message;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">                System.out.println(textMessage.getText());</span><br><span class="line">            &#125; <span class="keyword">catch</span> (JMSException e) &#123;</span><br><span class="line">                e.printStackTrace();</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h1 id="十、SpringBoot整合ActiveMQ"><a href="#十、SpringBoot整合ActiveMQ" class="headerlink" title="十、SpringBoot整合ActiveMQ"></a>十、SpringBoot整合ActiveMQ</h1><h2 id="1-队列生产者"><a href="#1-队列生产者" class="headerlink" title="1.队列生产者"></a>1.队列生产者</h2><h3 id="1-1-pom-xml导入依赖"><a href="#1-1-pom-xml导入依赖" class="headerlink" title="1.1 pom.xml导入依赖"></a>1.1 pom.xml导入依赖</h3><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework.boot<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-boot-starter-activemq<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.1.5.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h3 id="1-2-配置文件application-properties"><a href="#1-2-配置文件application-properties" class="headerlink" title="1.2 配置文件application.properties"></a>1.2 配置文件application.properties</h3><figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">server.port</span>=<span class="string">8888</span></span><br><span class="line"><span class="comment"># MQ服务器地址</span></span><br><span class="line"><span class="meta">spring.activemq.broker-url</span>=<span class="string">tcp://192.168.227.20:61616</span></span><br><span class="line"><span class="meta">spring.activemq.user</span>=<span class="string">admin</span></span><br><span class="line"><span class="meta">spring.activemq.password</span>=<span class="string">admin</span></span><br><span class="line"><span class="comment">#发布订阅，false表示目的地为队列，true表示主题，不写默认队列</span></span><br><span class="line"><span class="meta">spring.jms.pub-sub-domain</span>=<span class="string">false</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义队列名称</span></span><br><span class="line"><span class="attr">myqueue</span>=<span class="string">boot-activemq-queue</span></span><br></pre></td></tr></table></figure>

<h3 id="1-3-使用注解配置Bean"><a href="#1-3-使用注解配置Bean" class="headerlink" title="1.3 使用注解配置Bean"></a>1.3 使用注解配置Bean</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">ConfigBean</span> </span>&#123;</span><br><span class="line">    <span class="comment">//从配置文件中获取</span></span><br><span class="line">    <span class="meta">@Value(&quot;$&#123;myqueue&#125;&quot;)</span></span><br><span class="line">    <span class="keyword">private</span> String myqueue;</span><br><span class="line">    <span class="comment">//注入容器</span></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Queue <span class="title">queue</span><span class="params">()</span></span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> ActiveMQQueue(myqueue);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="1-4-业务层发送消息"><a href="#1-4-业务层发送消息" class="headerlink" title="1.4 业务层发送消息"></a>1.4 业务层发送消息</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@EnableJms</span>  <span class="comment">//开启基于注解的jms</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Queue_Produce</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> JmsMessagingTemplate jmsMessagingTemplate;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> Queue queue;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">produceMsg</span><span class="params">()</span></span>&#123;</span><br><span class="line">        jmsMessagingTemplate.convertAndSend(queue,<span class="string">&quot;====&quot;</span>+ UUID.randomUUID().toString().substring(<span class="number">0</span>,<span class="number">6</span>));</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="1-5-测试"><a href="#1-5-测试" class="headerlink" title="1.5 测试"></a>1.5 测试</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@SpringBootTest(classes = BootMqProducerApplication.class)</span></span><br><span class="line"><span class="meta">@RunWith(SpringJUnit4ClassRunner.class)</span></span><br><span class="line"><span class="meta">@WebAppConfiguration</span></span><br><span class="line"><span class="class"><span class="keyword">class</span> <span class="title">BootMqProducerApplicationTests</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Resource</span></span><br><span class="line">    <span class="keyword">private</span> Queue_Produce queue_produce;</span><br><span class="line">    <span class="meta">@Test</span></span><br><span class="line">    <span class="function"><span class="keyword">void</span> <span class="title">contextLoads</span><span class="params">()</span> </span>&#123;</span><br><span class="line">        queue_produce.produceMsg();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="1-6生产者间隔定投"><a href="#1-6生产者间隔定投" class="headerlink" title="1.6生产者间隔定投"></a>1.6生产者间隔定投</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br></pre></td><td class="code"><pre><span class="line"> <span class="comment">//间隔三秒定投</span></span><br><span class="line"><span class="meta">@Scheduled(fixedDelay = 3000)</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">produceMsgScheduled</span><span class="params">()</span></span>&#123;</span><br><span class="line">    jmsMessagingTemplate.convertAndSend(queue,<span class="string">&quot;====&quot;</span>+ UUID.randomUUID().toString().substring(<span class="number">0</span>,<span class="number">6</span>));</span><br><span class="line">    System.out.println(<span class="string">&quot;=====send over=====&quot;</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>然后在启动类上标记<font color='red'><strong>@EnableScheduling  开启基于注解的Scheduled</strong></font></p>
<p>接着启动主启动类，就可以实现定时发送消息的功能</p>
<h2 id="2-队列消费者"><a href="#2-队列消费者" class="headerlink" title="2.队列消费者"></a>2.队列消费者</h2><h3 id="2-1-pom-xml导入依赖"><a href="#2-1-pom-xml导入依赖" class="headerlink" title="2.1 pom.xml导入依赖"></a>2.1 pom.xml导入依赖</h3><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">dependency</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">groupId</span>&gt;</span>org.springframework.boot<span class="tag">&lt;/<span class="name">groupId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">artifactId</span>&gt;</span>spring-boot-starter-activemq<span class="tag">&lt;/<span class="name">artifactId</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">version</span>&gt;</span>2.1.5.RELEASE<span class="tag">&lt;/<span class="name">version</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">dependency</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h3 id="2-2-配置文件application-properties"><a href="#2-2-配置文件application-properties" class="headerlink" title="2.2 配置文件application.properties"></a>2.2 配置文件application.properties</h3><figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">server.port</span>=<span class="string">7777</span></span><br><span class="line"><span class="comment"># MQ服务器地址</span></span><br><span class="line"><span class="meta">spring.activemq.broker-url</span>=<span class="string">tcp://192.168.227.20:61616</span></span><br><span class="line"><span class="meta">spring.activemq.user</span>=<span class="string">admin</span></span><br><span class="line"><span class="meta">spring.activemq.password</span>=<span class="string">admin</span></span><br><span class="line"><span class="comment">#发布订阅，false表示目的地为队列，true表示主题，不写默认队列</span></span><br><span class="line"><span class="meta">spring.jms.pub-sub-domain</span>=<span class="string">false</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义队列名称</span></span><br><span class="line"><span class="attr">myqueue</span>=<span class="string">boot-activemq-queue</span></span><br></pre></td></tr></table></figure>

<h3 id="2-3-接收消息"><a href="#2-3-接收消息" class="headerlink" title="2.3 接收消息"></a>2.3 接收消息</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Queue_Consumer</span> </span>&#123;</span><br><span class="line">    <span class="meta">@JmsListener(destination = &quot;$&#123;myqueue&#125;&quot;)</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">receive</span><span class="params">(TextMessage textMessage)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        System.out.println(<span class="string">&quot;====消费者收到消息:&quot;</span>+textMessage.getText()+<span class="string">&quot;====&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h2 id="3-主题生产者"><a href="#3-主题生产者" class="headerlink" title="3.主题生产者"></a>3.主题生产者</h2><h3 id="3-1导入依赖"><a href="#3-1导入依赖" class="headerlink" title="3.1导入依赖"></a>3.1导入依赖</h3><h3 id="3-2配置文件"><a href="#3-2配置文件" class="headerlink" title="3.2配置文件"></a>3.2配置文件</h3><figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">server.port</span>=<span class="string">6666</span></span><br><span class="line"><span class="comment"># MQ服务器地址</span></span><br><span class="line"><span class="meta">spring.activemq.broker-url</span>=<span class="string">tcp://192.168.227.20:61616</span></span><br><span class="line"><span class="meta">spring.activemq.user</span>=<span class="string">admin</span></span><br><span class="line"><span class="meta">spring.activemq.password</span>=<span class="string">admin</span></span><br><span class="line"><span class="comment">#发布订阅，false表示目的地为队列，true表示主题，不写默认队列</span></span><br><span class="line"><span class="meta">spring.jms.pub-sub-domain</span>=<span class="string">true</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义队列名称</span></span><br><span class="line"><span class="attr">myqueue</span>=<span class="string">boot-activemq-queue</span></span><br><span class="line"><span class="comment">#自定义主题名称</span></span><br><span class="line"><span class="attr">myTopic</span>=<span class="string">boot-activemq-topic</span></span><br></pre></td></tr></table></figure>

<h3 id="3-3配置类"><a href="#3-3配置类" class="headerlink" title="3.3配置类"></a>3.3配置类</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Configuration</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">ConfigBean</span> </span>&#123;</span><br><span class="line">    <span class="comment">//从配置文件中获取</span></span><br><span class="line">    <span class="meta">@Value(&quot;$&#123;myTopic&#125;&quot;)</span></span><br><span class="line">    <span class="keyword">private</span> String myTopic;</span><br><span class="line">    <span class="comment">//注入容器</span></span><br><span class="line">    <span class="meta">@Bean</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> Topic <span class="title">topic</span><span class="params">()</span></span>&#123;</span><br><span class="line">        <span class="keyword">return</span> <span class="keyword">new</span> ActiveMQTopic(myTopic);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="3-4业务逻辑"><a href="#3-4业务逻辑" class="headerlink" title="3.4业务逻辑"></a>3.4业务逻辑</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="meta">@EnableJms</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Topic_Produce</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> JmsMessagingTemplate jmsMessagingTemplate;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    <span class="keyword">private</span> Topic topic;</span><br><span class="line"></span><br><span class="line">    <span class="meta">@Scheduled(fixedDelay = 3000)</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">produceTopic</span><span class="params">()</span></span>&#123;</span><br><span class="line">        jmsMessagingTemplate.convertAndSend(topic,<span class="string">&quot;主题消息:&quot;</span>+ UUID.randomUUID().toString().substring(<span class="number">0</span>,<span class="number">6</span>));</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h3 id="3-4主启动类标记-EnableScheduling"><a href="#3-4主启动类标记-EnableScheduling" class="headerlink" title="3.4主启动类标记@EnableScheduling"></a>3.4主启动类标记@EnableScheduling</h3><p>开启基于注解的Scheduled（定时发送消息）</p>
<h2 id="4-主题消费者"><a href="#4-主题消费者" class="headerlink" title="4.主题消费者"></a>4.主题消费者</h2><h3 id="4-1导入依赖"><a href="#4-1导入依赖" class="headerlink" title="4.1导入依赖"></a>4.1导入依赖</h3><h3 id="4-2配置文件"><a href="#4-2配置文件" class="headerlink" title="4.2配置文件"></a>4.2配置文件</h3><figure class="highlight properties"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">server.port</span>=<span class="string">5556</span></span><br><span class="line"><span class="comment"># MQ服务器地址</span></span><br><span class="line"><span class="meta">spring.activemq.broker-url</span>=<span class="string">tcp://192.168.227.20:61616</span></span><br><span class="line"><span class="meta">spring.activemq.user</span>=<span class="string">admin</span></span><br><span class="line"><span class="meta">spring.activemq.password</span>=<span class="string">admin</span></span><br><span class="line"><span class="comment">#发布订阅，false表示目的地为队列，true表示主题，不写默认队列</span></span><br><span class="line"><span class="meta">spring.jms.pub-sub-domain</span>=<span class="string">true</span></span><br><span class="line"><span class="comment"></span></span><br><span class="line"><span class="comment">#定义队列名称</span></span><br><span class="line"><span class="attr">myqueue</span>=<span class="string">boot-activemq-queue</span></span><br><span class="line"><span class="comment">#自定义主题名称</span></span><br><span class="line"><span class="attr">myTopic</span>=<span class="string">boot-activemq-topic</span></span><br></pre></td></tr></table></figure>

<h3 id="4-3业务逻辑"><a href="#4-3业务逻辑" class="headerlink" title="4.3业务逻辑"></a>4.3业务逻辑</h3><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Topic_Consumer</span> </span>&#123;</span><br><span class="line">    <span class="meta">@JmsListener(destination = &quot;$&#123;myTopic&#125;&quot;)</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">receive</span><span class="params">(TextMessage textMessage)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        System.out.println(<span class="string">&quot;====消费者收到订阅的消息:&quot;</span>+textMessage.getText());</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h1 id="十一、ActiveMQ的传输协议"><a href="#十一、ActiveMQ的传输协议" class="headerlink" title="十一、ActiveMQ的传输协议"></a>十一、ActiveMQ的传输协议</h1><h2 id="1-是什么"><a href="#1-是什么" class="headerlink" title="1.是什么"></a>1.是什么</h2><p>ActiveMQ支持的client-broker通讯协议有：TCP、NIO、UDP、SSL、Http(s)、VM。</p>
<p>其中配置Transport Connector的文件在activeMQ安装目录的conf/activemq.xml中的<transportConnectors>标签之内。</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112192322109.png" alt="image-20211219232229232"></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112192323564.png" alt="image-20211219232350722"></p>
<p>在上文的配置信息中，</p>
<p>URI描述信息的头部都是采用协议名称:例如：</p>
<p>描述amqp协议的监听端口时，采用的URI描述格式为“amqp://….”</p>
<p>描述stomp协议的监听端口时，采用的URI描述格式为“stomp://….”</p>
<p>唯独在进行openwire协议描述时，URI头却采用”tcp://…”。<font color='red'>这是因为ActiveMQ中默认的消息协议就是openwire</font></p>
<h2 id="2-有哪些协议"><a href="#2-有哪些协议" class="headerlink" title="2.有哪些协议"></a>2.有哪些协议</h2><h3 id="2-1-Transmission-Control-Protocol-TCP-默认"><a href="#2-1-Transmission-Control-Protocol-TCP-默认" class="headerlink" title="2.1 Transmission Control Protocol(TCP)默认"></a>2.1 Transmission Control Protocol(TCP)默认</h3><p>属性查看官方文档：<a target="_blank" rel="noopener" href="https://activemq.apache.org/tcp-transport-reference">https://activemq.apache.org/tcp-transport-reference</a></p>
<h3 id="2-2-New-I-O-API-Protocol-NIO"><a href="#2-2-New-I-O-API-Protocol-NIO" class="headerlink" title="2.2 New I/O API Protocol(NIO)"></a>2.2 New I/O API Protocol(NIO)</h3><p>1）NIO协议和TCP协议类似但NIO更侧重于底层得访问操作。比TCP提供更好得性能</p>
<p>2）NIO连接得URI形式：nio://hostname:port?key=value</p>
<p>3）Transport Connector配置实例，参考官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/configuring-transports">https://activemq.apache.org/configuring-transports</a></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112202207155.png" alt="image-20211220220731694"></p>
<h3 id="2-3-AMQP协议"><a href="#2-3-AMQP协议" class="headerlink" title="2.3 AMQP协议"></a>2.3 AMQP协议</h3><h3 id="2-4-stomp协议"><a href="#2-4-stomp协议" class="headerlink" title="2.4 stomp协议"></a>2.4 stomp协议</h3><h3 id="2-5-Secure-Sockets-Layer-Protocol-SSL"><a href="#2-5-Secure-Sockets-Layer-Protocol-SSL" class="headerlink" title="2.5 Secure Sockets Layer Protocol(SSL)"></a>2.5 Secure Sockets Layer Protocol(SSL)</h3><h3 id="2-6-mqtt协议"><a href="#2-6-mqtt协议" class="headerlink" title="2.6 mqtt协议"></a>2.6 mqtt协议</h3><h3 id="2-7ws协议"><a href="#2-7ws协议" class="headerlink" title="2.7ws协议"></a>2.7ws协议</h3><h3 id="2-8总结"><a href="#2-8总结" class="headerlink" title="2.8总结"></a>2.8总结</h3><table>
<thead>
<tr>
<th>协议</th>
<th>概述</th>
</tr>
</thead>
<tbody><tr>
<td>TCP</td>
<td>默认的协议，性能相对可以</td>
</tr>
<tr>
<td>NIO</td>
<td>基于TCP协议之上的，进行了扩展和优化，具有更好的扩展性</td>
</tr>
<tr>
<td>UDP</td>
<td>性能比TCP更好，但是不具有可靠性</td>
</tr>
<tr>
<td>SSL</td>
<td>安全连接</td>
</tr>
<tr>
<td>HTTP（S）</td>
<td>基于HTTP或者HTTPS</td>
</tr>
<tr>
<td>VM</td>
<td>VM本省不是协议，当客户端和代理再同一个Java虚拟机（VM）中运行时，他们之间需要通信，但不想占用网络通道，而是直接通信，可以使用该方式</td>
</tr>
</tbody></table>
<h2 id="3-NIO案例演示"><a href="#3-NIO案例演示" class="headerlink" title="3.NIO案例演示"></a>3.NIO案例演示</h2><p>修改配置文件</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">transportConnectors</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">transportConnector</span> <span class="attr">name</span>=<span class="string">&quot;nio&quot;</span> <span class="attr">uri</span>=<span class="string">&quot;nio://0.0.0.0:61618?trace=true&quot;</span>/&gt;</span>  </span><br><span class="line">&lt;/<span class="tag">&lt;<span class="name">transportConnectors</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="4-NIO案例演示增强"><a href="#4-NIO案例演示增强" class="headerlink" title="4.NIO案例演示增强"></a>4.NIO案例演示增强</h2><blockquote>
<p>如何让一个端口技能支持NIO网络IO模型，又让它支持多个协议</p>
</blockquote>
<p>NIO+TCP/stomp/…</p>
<p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/auto">https://activemq.apache.org/auto</a></p>
<p><strong>解决：</strong></p>
<p>使用auto关键字</p>
<p>使用“+”符号来为端口设置多个特性</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">transportConnector</span> <span class="attr">name</span>=<span class="string">&quot;auto+nio&quot;</span> <span class="attr">uri</span>=<span class="string">&quot;auto+nio://0.0.0.0:61608?maximumConnections=1000<span class="symbol">&amp;amp;</span>wireFormat.maxFrameSize=104857600<span class="symbol">&amp;amp;</span>org.apache.activemq.transport.nio.SelectorManager.corePoolSize=20<span class="symbol">&amp;amp;</span>org.activemq.transport.nio.SelectorManager.maximumPoolSize=50&quot;</span>/&gt;</span></span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112210007281.png" alt="image-20211221000658392"></p>
<p>四种协议会被自动检测到</p>
<h1 id="十二、ActiveMQ的消息存储和可持久化"><a href="#十二、ActiveMQ的消息存储和可持久化" class="headerlink" title="十二、ActiveMQ的消息存储和可持久化"></a>十二、ActiveMQ的消息存储和可持久化</h1><p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112211658815.png" alt="image-20211221165821373"></p>
<h2 id="1-AMQ持久化机制（面试题）"><a href="#1-AMQ持久化机制（面试题）" class="headerlink" title="1.AMQ持久化机制（面试题）"></a>1.AMQ持久化机制（面试题）</h2><p>replicated levelDB、KahaDB</p>
<p><strong>说明：</strong></p>
<p>为了避免意外宕机以后丢失信息，需要做到重启后可以恢复消息队列，消息系统一般会采用持久化机制</p>
<p>ActiveMQ的消息持久化机制有JDBC，AMQ，KahaDB和LevelDB，无论使用哪种，消息的存储逻辑都是一致的。</p>
<p>当发送者将消息发送后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等再试图将消息发送给接收者，<font color='red'>成功则将消息从存储中删除</font>，失败则继续尝试发送</p>
<h2 id="2-有哪些"><a href="#2-有哪些" class="headerlink" title="2.有哪些"></a>2.有哪些</h2><blockquote>
<p>AMQ Message Store（了解）</p>
</blockquote>
<p>基于文件的存储方式，是以前的默认消息存储，现在不用了</p>
<blockquote>
<p>Kaha DB消息存储（默认）</p>
</blockquote>
<p>基于日志文件，从ActiveMQ5.4开始默认的持久化插件</p>
<p>在activemq.xml文件中，持节化的信息会放到指定的目录下</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112211615402.png" alt="image-20211221161528713"></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112211616363.png" alt="image-20211221161629843"></p>
<p>KahaDB是目前默认的存储方式，可用于任何场景，提高性能和恢复能力</p>
<p><font color='red'>消息存储使用一个</font><font color='blue'>事务日志</font><font color='red'>和仅仅用一个索引文件来存储它所有的地址</font></p>
<p>数据会被追加到data log中。当不在需要log文件中的数据的时候，log文件会被丢弃</p>
<p>db.data存储BTree索引信息</p>
<p>db.redo用于恢复</p>
<blockquote>
<p>JDBC消息存储</p>
</blockquote>
<p>消息基于JDBC存储</p>
<blockquote>
<p>LevelDB 消息存储</p>
</blockquote>
<p>这种文件系统是ActiveMQ5.8之后引进的，它和KahaDB非常相似，也是基于文件的本地数据存储信息，但是它提供比KahaDB更快的持久性</p>
<p>但它不使用自定义B-Tree实现来索引预写日志，而是使用基于LevelDB的所有</p>
<blockquote>
<p>JDBC Message store with ActiveMQ Journal</p>
</blockquote>
<h2 id="3-JDBC消息存储"><a href="#3-JDBC消息存储" class="headerlink" title="3.JDBC消息存储"></a>3.JDBC消息存储</h2><p>1.MQ+MySQL</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112230916353.png" alt="image-20211223091636554"></p>
<p>2.添加mysql数据库的驱动包到lib文件夹（mysql-connector-java-5.1.30.jar）</p>
<p>3.jdbcPersistenceAdapter配置</p>
<p>在/myactiveMQ/apache-activemq-5.15.9/conf路径下修改activemq.xml配置文件：</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112230929596.png" alt="image-20211223092949103"></p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">persistenceAdapter</span>&gt;</span> </span><br><span class="line">  <span class="tag">&lt;<span class="name">jdbcPersistenceAdapter</span> <span class="attr">dataSource</span>=<span class="string">&quot;#mysql-ds&quot;</span>/&gt;</span> </span><br><span class="line"><span class="tag">&lt;/<span class="name">persistenceAdapter</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>dataSource指定将要引用的持久化数据库的bean名称</p>
<p>createTablesOnStartUp是否在启动的时候创建数据表，默认值是true，这样每次启动都会创建数据表，一般是第一次启动的时候设置为true之后改成false</p>
<h2 id="4-数据库连接池配置"><a href="#4-数据库连接池配置" class="headerlink" title="4.数据库连接池配置"></a>4.数据库连接池配置</h2><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">bean</span> <span class="attr">id</span>=<span class="string">&quot;mysql-ds&quot;</span> <span class="attr">class</span>=<span class="string">&quot;org.apache.commons.dbcp2.BasicDataSource&quot;</span> <span class="attr">destroy-method</span>=<span class="string">&quot;close&quot;</span>&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;driverClassName&quot;</span> <span class="attr">value</span>=<span class="string">&quot;com.mysql.jdbc.Driver&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;url&quot;</span> <span class="attr">value</span>=<span class="string">&quot;jdbc:mysql://192.168.43.228:3306/activemq?relaxAutoCommit=true&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;username&quot;</span> <span class="attr">value</span>=<span class="string">&quot;root&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;password&quot;</span> <span class="attr">value</span>=<span class="string">&quot;123456&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;maxTotal&quot;</span> <span class="attr">value</span>=<span class="string">&quot;200&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;<span class="name">property</span> <span class="attr">name</span>=<span class="string">&quot;poolPreparedStatements&quot;</span> <span class="attr">value</span>=<span class="string">&quot;true&quot;</span>/&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">bean</span>&gt;</span></span><br></pre></td></tr></table></figure>

<p>需要粘贴在broker和import两个标签之间</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112230956441.png" alt="image-20211223095625347"></p>
<h2 id="5-sql建仓和建表"><a href="#5-sql建仓和建表" class="headerlink" title="5.sql建仓和建表"></a>5.sql建仓和建表</h2><figure class="highlight sql"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">create</span> database activemq</span><br></pre></td></tr></table></figure>

<p>三张表的说明</p>
<ul>
<li>ACTIVEMQ_MSGS：消息</li>
<li>ACTIVEMQ_ACKS：签收</li>
<li>ACTIVEMQ_LOCK：锁定</li>
</ul>
<h2 id="6-代码验证"><a href="#6-代码验证" class="headerlink" title="6.代码验证"></a>6.代码验证</h2><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">//需要开启持久化</span></span><br><span class="line">messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);</span><br></pre></td></tr></table></figure>

<h2 id="7-数据库情况"><a href="#7-数据库情况" class="headerlink" title="7.数据库情况"></a>7.数据库情况</h2><p>1.一旦运行生产code</p>
<ul>
<li><strong>点对点情况：</strong></li>
</ul>
<p>当DeliveryMode设置为NON——PERSISTENCE时，消息被保存在内存中；</p>
<p>当DeliveryMode设置为PERSISTENCE时，消息保存在broker的相应的文件或者数据库中。</p>
<p><strong>而且点对点的消息一旦被消费就从broker中消除</strong></p>
<ul>
<li><strong>发布订阅情况：</strong></li>
</ul>
<p>当DeliveryMode设置为PERSISTENCE时，</p>
<p>在数据库的activemq_acks会<strong>保存订阅者信息</strong></p>
<p>在activemq_msgs会<strong>保存发布的消息</strong></p>
<h2 id="8-开发常见问题"><a href="#8-开发常见问题" class="headerlink" title="8.开发常见问题"></a>8.开发常见问题</h2><h3 id="数据库jar包"><a href="#数据库jar包" class="headerlink" title="数据库jar包"></a><font color='blue'>数据库jar包</font></h3><p>需要注意使用到的jar文件放置到ActiveMQ安装路径下的lib目录。mysql-jdbc驱动的jar包和对于的数据库连接池jar包</p>
<h3 id="createTablesOnStartup属性"><a href="#createTablesOnStartup属性" class="headerlink" title="createTablesOnStartup属性"></a><font color='blue'>createTablesOnStartup属性</font></h3><p>在jdbcPersistenceAdapter标签中设置了createTablesOnStartup属性为true时，在第一次启动ActiveMQ时，ActiveMQ服务节点会自动创建所需要的数据表，启动完成后可以去掉这个属性，或者更改createTablesOnStartup属性为false</p>
<h3 id="下划线问题"><a href="#下划线问题" class="headerlink" title="下划线问题"></a><font color='blue'>下划线问题</font></h3><p>“java.lang.IllegalStateException:BeanFactory not initialized or already closed”</p>
<p>这是因为操作系统的机器名中有”_”符号。将机器名更改后重启即可解决问题</p>
<h2 id="9-JDBC-Message-store-with-ActiveMQ-Journal"><a href="#9-JDBC-Message-store-with-ActiveMQ-Journal" class="headerlink" title="9.JDBC Message store with ActiveMQ Journal"></a>9.JDBC Message store with ActiveMQ Journal</h2><h3 id="9-1说明"><a href="#9-1说明" class="headerlink" title="9.1说明"></a>9.1说明</h3><ul>
<li><p>这种方式克服了JDBC Store的不足，JDBC每次消息过来，都需要去写库和读库</p>
</li>
<li><p>ActiveMQ Journal，使用高速缓存写入技术，大大提高性能</p>
</li>
<li><p>当消费者的消费速度能够及时跟上生产者消息的生产速度时，journal文件能够大大减少需要写入到DB中的消息</p>
</li>
<li><p>如果消费者的消费速度很慢，这个时候journal文件可以使消息以批量方式写道DB。</p>
</li>
</ul>
<h3 id="9-2配置"><a href="#9-2配置" class="headerlink" title="9.2配置"></a>9.2配置</h3><figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">persistenceFactory</span>&gt;</span></span><br><span class="line">	<span class="tag">&lt;<span class="name">journalPersistenceAdapterFactory</span> <span class="attr">journalLogFiles</span>=<span class="string">&quot;5&quot;</span> </span></span><br><span class="line"><span class="tag">                                      <span class="attr">journalLogFileSize</span>=<span class="string">&quot;32768&quot;</span></span></span><br><span class="line"><span class="tag">                                      <span class="attr">useJournal</span>=<span class="string">&quot;true&quot;</span></span></span><br><span class="line"><span class="tag">                                      <span class="attr">useQuickJournal</span>=<span class="string">&quot;true&quot;</span></span></span><br><span class="line"><span class="tag">                                      <span class="attr">dataSource</span>=<span class="string">&quot;#mysql-ds&quot;</span></span></span><br><span class="line"><span class="tag">                                      <span class="attr">dataDirectory</span>=<span class="string">&quot;activemq-data&quot;</span> /&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">persistenceFactory</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h3 id="9-3重启激活"><a href="#9-3重启激活" class="headerlink" title="9.3重启激活"></a>9.3重启激活</h3><h2 id="10-ActiveMQ持久化机制总结"><a href="#10-ActiveMQ持久化机制总结" class="headerlink" title="10.ActiveMQ持久化机制总结"></a>10.ActiveMQ持久化机制总结</h2><h3 id="持久化消息主要是指："><a href="#持久化消息主要是指：" class="headerlink" title="持久化消息主要是指："></a>持久化消息主要是指：</h3><p>MQ所在的服务器down了消息不会丢失的机制</p>
<h3 id="持久化机制演化过程："><a href="#持久化机制演化过程：" class="headerlink" title="持久化机制演化过程："></a>持久化机制演化过程：</h3><ul>
<li>从最初的AMQ Message Store方案到ActiveMQ V4版本中推出的High performance journal（高性能事务支持），并且同步推出了关于关系型数据库的存储方案。</li>
<li>ActiveMQ 5.3版本中又推出对KahaDB的支持（V5.4之后被称为默认的持久化方案）</li>
<li>后来ActiveMQ V5.8开始支持LevelDB</li>
<li>V5.9+版本提高推出的Zookeeper+LevelDB集群化方案。</li>
<li>现V5.14.2版本是LevelDB弃用公告的第一个版本。</li>
<li>V5.17.0版本删除了LevelDb持久化机制</li>
</ul>
<h3 id="ActiveMQ的消息持久化机制有："><a href="#ActiveMQ的消息持久化机制有：" class="headerlink" title="ActiveMQ的消息持久化机制有："></a>ActiveMQ的消息持久化机制有：</h3><p>AMQ：基于日志文件</p>
<p>KahaDB：基于日志文件，5.4版本后成为默认的持久化插件</p>
<p>JDBC：基于第三方数据库</p>
<p>LevelDB：基于文件的本地数据库存储</p>
<p>Replicated LevelDB Store：ActiveMQ 5.9提供基于LevelDB和Zookeeper的数据赋值方式，用于Master-slave方式的首选数据复制方案</p>
<h1 id="十三、ActiveMQ多节点集群"><a href="#十三、ActiveMQ多节点集群" class="headerlink" title="十三、ActiveMQ多节点集群"></a>十三、ActiveMQ多节点集群</h1><h2 id="面试题"><a href="#面试题" class="headerlink" title="面试题"></a>面试题</h2><p>引入消息队列之后该如何保证其高可用性</p>
<h2 id="是什么"><a href="#是什么" class="headerlink" title="是什么"></a>是什么</h2><p>基于Zookeeper和LevelDB搭建ActiveMQ集群。</p>
<p>集群仅提供主从方式的高可用集群功能，避免单点故障</p>
<h2 id="Zookeeper-Replicated-LevelDB-Store的主从集群"><a href="#Zookeeper-Replicated-LevelDB-Store的主从集群" class="headerlink" title="Zookeeper+Replicated LevelDB Store的主从集群"></a>Zookeeper+Replicated LevelDB Store的主从集群</h2><p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/replicated-leveldb-store">https://activemq.apache.org/replicated-leveldb-store</a></p>
<h3 id="前提条件"><a href="#前提条件" class="headerlink" title="前提条件"></a>前提条件</h3><p>搭建好Zookeeper的集群，可参考Zookeeper的笔记</p>
<h3 id="集群部署规划表"><a href="#集群部署规划表" class="headerlink" title="集群部署规划表"></a>集群部署规划表</h3><table>
<thead>
<tr>
<th>主机</th>
<th>Zookeeper集群端口</th>
<th>AMQ集群bind端口</th>
<th>AMQ消息tcp端口</th>
<th>管理控制台端口</th>
<th>AMQ节点安装目录</th>
</tr>
</thead>
<tbody><tr>
<td>192.168.227.20</td>
<td>2184</td>
<td>bind=”tcp://0.0.0.0:63631”</td>
<td>61616</td>
<td>8161</td>
<td>/mq_cluster/mq_node01</td>
</tr>
<tr>
<td>192.168.227.20</td>
<td>2182</td>
<td>bind=”tcp://0.0.0.0:63632”</td>
<td>61617</td>
<td>8162</td>
<td>/mq_cluster/mq_node02</td>
</tr>
<tr>
<td>192.168.227.20</td>
<td>2183</td>
<td>bind=”tcp://0.0.0.0:63633”</td>
<td>61618</td>
<td>8163</td>
<td>/mq_cluster/mq_node03</td>
</tr>
</tbody></table>
<h3 id="创建三台集群目录"><a href="#创建三台集群目录" class="headerlink" title="创建三台集群目录"></a>创建三台集群目录</h3><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">mkdir /mq_cluster/</span><br><span class="line"><span class="built_in">cd</span> /mq_cluster/</span><br><span class="line">cp -r /myactivemq/apache-activemq-5.15.9 mq_node01</span><br><span class="line">cp -r mq_node01 mq_node02</span><br><span class="line">cp -r mq_node01 mq_node03</span><br></pre></td></tr></table></figure>

<h3 id="修改管理控制台的端口"><a href="#修改管理控制台的端口" class="headerlink" title="修改管理控制台的端口"></a>修改管理控制台的端口</h3><p>mq_node01全部默认不动</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112242356204.png" alt="image-20211224235624110"></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112242357956.png" alt="image-20211224235727725"></p>
<h3 id="hostname名字映射"><a href="#hostname名字映射" class="headerlink" title="hostname名字映射"></a>hostname名字映射</h3><figure class="highlight plaintext"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">vim /etc/hosts</span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112250003022.png" alt="image-20211225000318743"></p>
<p>红框部分修改为自己的IP和地址映射</p>
<h3 id="ActiveMQ集群配置"><a href="#ActiveMQ集群配置" class="headerlink" title="ActiveMQ集群配置"></a>ActiveMQ集群配置</h3><p>三个节点的BrokerName需要一致（activemq.xml文件中）</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112250008461.png" alt="image-20211225000800291"></p>
<p>三个节点的持久化配置</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">broker</span> <span class="attr">brokerName</span>=<span class="string">&quot;broker&quot;</span> <span class="attr">...</span> &gt;</span></span><br><span class="line">  ...</span><br><span class="line">  <span class="tag">&lt;<span class="name">persistenceAdapter</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">replicatedLevelDB</span> <span class="attr">directory</span>=<span class="string">&quot;$&#123;activemq.data&#125;/leveldb&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">replicas</span>=<span class="string">&quot;3&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">bind</span>=<span class="string">&quot;tcp://0.0.0.0:63631&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">zkAddress</span>=<span class="string">&quot;localhost:2184,localhost:2182,localhost:2183&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">sync</span>=<span class="string">&quot;local_disk&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">zkPath</span>=<span class="string">&quot;/activemq/leveldb-stores&quot;</span></span></span><br><span class="line"><span class="tag">                       <span class="attr">hostname</span>=<span class="string">&quot;yuanmq-server&quot;</span>/&gt;</span></span><br><span class="line">  <span class="tag">&lt;/<span class="name">persistenceAdapter</span>&gt;</span></span><br><span class="line">  ...</span><br><span class="line"><span class="tag">&lt;/<span class="name">broker</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="修改各个节点的消息端口"><a href="#修改各个节点的消息端口" class="headerlink" title="修改各个节点的消息端口"></a>修改各个节点的消息端口</h2><p>在activemq.xml配置文件中修改</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112250022330.png" alt="image-20211225002233819"></p>
<h2 id="按顺序启动3个ActiveMQ节点，前提zk集群已经成功启动运行"><a href="#按顺序启动3个ActiveMQ节点，前提zk集群已经成功启动运行" class="headerlink" title="按顺序启动3个ActiveMQ节点，前提zk集群已经成功启动运行"></a>按顺序启动3个ActiveMQ节点，前提zk集群已经成功启动运行</h2><h2 id="连接集群"><a href="#连接集群" class="headerlink" title="连接集群"></a>连接集群</h2><figure class="highlight bash"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="built_in">cd</span> /usr/<span class="built_in">local</span>/zookeeper/apache-zookeeper-3.7.0-bin/bin</span><br><span class="line"></span><br><span class="line">./zkCli.sh -server 127.0.0.1:2184</span><br></pre></td></tr></table></figure>

<p>查看master</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112250112815.png" alt="image-20211225011226528"></p>
<p>elected选中，表示这台机器为主机。null表示从机</p>
<h2 id="集群可用性测试"><a href="#集群可用性测试" class="headerlink" title="集群可用性测试"></a>集群可用性测试</h2><p>ActiveMQ的客户端只能访问Master的Broker，其他处于Slave的Broker不能访问，所以客户端连接的Broker应该使用failover协议（失败转移）</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112250118002.png" alt="image-20211225011801744"></p>
<p>如果一个ActiveMQ1节点挂掉或者一个Zookeeper节点挂带哦，ActiveMQ服务依然正常运转，如果仅剩一个ActiveMQ节点，由于不能选举Master，所以ActiveMQ不能正常运行</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;failover:(tcp://192.168.227.20:61616,tcp://192.168.227.20:61617,tcp://192.168.227.20:61618)?randomize=false&quot;</span>;</span><br><span class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue-cluster&quot;</span>;</span><br></pre></td></tr></table></figure>

<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112251152209.png" alt="image-20211225115239374"></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112251154505.png" alt="image-20211225115410066"></p>
<h1 id="十四、高级特性和大厂常考重点"><a href="#十四、高级特性和大厂常考重点" class="headerlink" title="十四、高级特性和大厂常考重点"></a>十四、高级特性和大厂常考重点</h1><h2 id="1-引入消息队列之后该如何保证其高可用性"><a href="#1-引入消息队列之后该如何保证其高可用性" class="headerlink" title="1.引入消息队列之后该如何保证其高可用性"></a>1.引入消息队列之后该如何保证其高可用性</h2><p>持久化，事务，消息的签收，zookeeper+replicateed-levedb-store的主从集群</p>
<h2 id="2-异步投递Async-Sends"><a href="#2-异步投递Async-Sends" class="headerlink" title="2.异步投递Async Sends"></a>2.异步投递Async Sends</h2><p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/async-sends">https://activemq.apache.org/async-sends</a></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112251426932.png" alt="image-20211225142602865"></p>
<p>第一种开启异步投递的方法</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">new</span> ActiveMQConnectionFactory(<span class="string">&quot;tcp://locahost:61616?jms.useAsyncSend=true&quot;</span>);</span><br></pre></td></tr></table></figure>

<p>第二种</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(<span class="keyword">true</span>);</span><br></pre></td></tr></table></figure>

<p>第三种</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">((ActiveMQConnection)connection).setUseAsyncSend(<span class="keyword">true</span>);</span><br></pre></td></tr></table></figure>

<h3 id="异步发送如何确认发送成功——通过接收回执确认"><a href="#异步发送如何确认发送成功——通过接收回执确认" class="headerlink" title="异步发送如何确认发送成功——通过接收回执确认"></a><font color='red'>异步发送如何确认发送成功</font>——<font color='cornflowerblue'>通过接收回执确认</font></h3><p>异步发送丢失消息的场景是：生产者设置userAsyncSend=true，使用producer.send(msg)持续发送消息。</p>
<p>由于消息不阻塞，生产者会认为所有send的消息均被成功发送至MQ。</p>
<p>如果MQ突然宕机，此时生产者端内存中尚未被发送至MQ的消息<font color='red'>都会丢失</font></p>
<p>所以，正常的异步发送方法是<font color='red'>需要接收回调的。</font></p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line">ActiveMQMessageProducer messageProducer = (ActiveMQMessageProducer) session.createProducer(queue);</span><br><span class="line"></span><br><span class="line"> activeMQMessageProducer.send(textMessage, <span class="keyword">new</span> AsyncCallback() &#123;</span><br><span class="line">                <span class="meta">@Override</span></span><br><span class="line">                <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onSuccess</span><span class="params">()</span> </span>&#123;</span><br><span class="line">                    System.out.println(msgID + <span class="string">&quot;send success&quot;</span>);</span><br><span class="line">                &#125;</span><br><span class="line"></span><br><span class="line">                <span class="meta">@Override</span></span><br><span class="line">                <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">onException</span><span class="params">(JMSException e)</span> </span>&#123;</span><br><span class="line">                    System.err.println(msgID + <span class="string">&quot;send fail&quot;</span>);</span><br><span class="line">                &#125;</span><br><span class="line">            &#125;);</span><br></pre></td></tr></table></figure>

<h2 id="3-延时发送和定时发送"><a href="#3-延时发送和定时发送" class="headerlink" title="3.延时发送和定时发送"></a>3.延时发送和定时发送</h2><p>回顾linux：定时投递命令 vmstat -n 2 3</p>
<p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/delay-and-schedule-message-delivery">https://activemq.apache.org/delay-and-schedule-message-delivery</a></p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112251503238.png" alt="image-20211225150259761"></p>
<p><strong>四大属性</strong></p>
<table>
<thead>
<tr>
<th>Property name</th>
<th>type</th>
<th>description</th>
</tr>
</thead>
<tbody><tr>
<td>AMQ_SCHEDULED_DELAY</td>
<td>long</td>
<td>延迟投递的时间（以毫秒为单位）</td>
</tr>
<tr>
<td>AMQ_SCHEDULED_PERIOD</td>
<td>long</td>
<td>重复投递的时间间隔（以毫秒为单位）</td>
</tr>
<tr>
<td>AMQ_SCHEDULED_REPEAT</td>
<td>int</td>
<td>重复投递的次数</td>
</tr>
<tr>
<td>AMQ_SCHEDULED_CRON</td>
<td>String</td>
<td>使用 Cron 表达式</td>
</tr>
</tbody></table>
<p>cron官网：<a target="_blank" rel="noopener" href="https://en.wikipedia.org/wiki/Cron">https://en.wikipedia.org/wiki/Cron</a></p>
<p>步骤：</p>
<p>1.在activemq.xml配置文件中给broker设置属性，表示开启延时投递和定时投递</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line">schedulerSupport=&quot;true&quot;</span><br></pre></td></tr></table></figure>

<p>2.java代码种封装的辅助消息类型：ScheduledMessage</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br><span class="line">73</span><br><span class="line">74</span><br><span class="line">75</span><br></pre></td><td class="code"><pre><span class="line">==============生产者=============</span><br><span class="line"></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsProducer_DelayAndSchedule</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue-delay&quot;</span>;</span><br><span class="line"></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> JMSException </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//5.创建消息的生产者</span></span><br><span class="line">        MessageProducer messageProducer = session.createProducer(queue);</span><br><span class="line">        <span class="keyword">long</span> delay = <span class="number">3000</span>;</span><br><span class="line">        <span class="keyword">long</span> period = <span class="number">4000</span>;</span><br><span class="line">        <span class="keyword">int</span> repeat = <span class="number">5</span>; <span class="comment">//重读投递的次数为5</span></span><br><span class="line">        <span class="comment">//设置持久化</span></span><br><span class="line">        messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);</span><br><span class="line">        <span class="comment">//6.通过使用MEssageProducer生产三条消息发送到MQ的队列里面</span></span><br><span class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">1</span>; i &lt;= <span class="number">3</span>; i++) &#123;</span><br><span class="line">            <span class="comment">//7.创建消息</span></span><br><span class="line">            TextMessage textMessage = session.createTextMessage(<span class="string">&quot;delay message---&quot;</span> + i);</span><br><span class="line"></span><br><span class="line">            textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,delay);</span><br><span class="line">            textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD,period);</span><br><span class="line">            textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,repeat);</span><br><span class="line"></span><br><span class="line">            <span class="comment">//8.通过messageProducer发送给mq</span></span><br><span class="line">            messageProducer.send(textMessage);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="comment">//9.关闭资源</span></span><br><span class="line">        messageProducer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">        System.out.println(<span class="string">&quot;=====消息发布到MQ完成=====&quot;</span>);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line">=================消费者==================</span><br><span class="line"></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">JmsConsumer_DelayAndSchedule</span> </span>&#123;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String ACTIVE_URL = <span class="string">&quot;tcp://192.168.227.20:61616&quot;</span>;</span><br><span class="line">    <span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> String QUEUE_NAME = <span class="string">&quot;queue-delay&quot;</span>;</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">        <span class="comment">//1.创建连接工厂</span></span><br><span class="line">        ActiveMQConnectionFactory activeMQConnectionFactory = <span class="keyword">new</span> ActiveMQConnectionFactory(ACTIVE_URL);</span><br><span class="line">        <span class="comment">//2.获取连接connection</span></span><br><span class="line">        Connection connection = activeMQConnectionFactory.createConnection();</span><br><span class="line">        connection.start();</span><br><span class="line">        <span class="comment">//3.创建session</span></span><br><span class="line">        <span class="comment">//两个参数：第一个叫事务，第二个叫签收</span></span><br><span class="line">        Session session = connection.createSession(<span class="keyword">false</span>, Session.AUTO_ACKNOWLEDGE);</span><br><span class="line">        <span class="comment">//4.创建目的地</span></span><br><span class="line">        Queue queue = session.createQueue(QUEUE_NAME);</span><br><span class="line">        <span class="comment">//创建消费者</span></span><br><span class="line">        MessageConsumer consumer = session.createConsumer(queue);</span><br><span class="line">        <span class="keyword">while</span>(<span class="keyword">true</span>)&#123;</span><br><span class="line">            TextMessage textMessage = (TextMessage) consumer.receive();</span><br><span class="line">            <span class="keyword">if</span>(textMessage != <span class="keyword">null</span>)&#123;</span><br><span class="line">                System.out.println(<span class="string">&quot;=====消费者接收到消息:&quot;</span>+textMessage.getText()+<span class="string">&quot;=====&quot;</span>);</span><br><span class="line">            &#125;<span class="keyword">else</span>&#123;</span><br><span class="line">                <span class="keyword">break</span>;</span><br><span class="line">            &#125;</span><br><span class="line">        &#125;</span><br><span class="line">        consumer.close();</span><br><span class="line">        session.close();</span><br><span class="line">        connection.close();</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h2 id="4-消费者重试机制"><a href="#4-消费者重试机制" class="headerlink" title="4.消费者重试机制"></a>4.消费者重试机制</h2><p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/redelivery-policy">https://activemq.apache.org/redelivery-policy</a></p>
<h3 id="具体哪些情况会引起消息重发"><a href="#具体哪些情况会引起消息重发" class="headerlink" title="具体哪些情况会引起消息重发"></a>具体哪些情况会引起消息重发</h3><ol>
<li>生产者开启了事务，且session中调用了rollback()</li>
<li>生产者开启事务且在调用commit()之前关闭或者没有commit</li>
<li>生产者在CLIENT_ACKNOWLEDGE的传递模式下，在session中调用了recover()</li>
<li>客户端连接超时（可能正在执行的代码比配置的超时时间长）。</li>
</ol>
<h3 id="请说说消息默认的重发时间间隔和重回发次数"><a href="#请说说消息默认的重发时间间隔和重回发次数" class="headerlink" title="请说说消息默认的重发时间间隔和重回发次数"></a>请说说消息默认的重发时间间隔和重回发次数</h3><p>间隔：1</p>
<p>次数：6</p>
<h3 id="有毒消息Poison-ACK的理解"><a href="#有毒消息Poison-ACK的理解" class="headerlink" title="有毒消息Poison ACK的理解"></a>有毒消息Poison ACK的理解</h3><p>一个消息被redelivedred超过默认的最大重发次数时，消费端会给MQ发送一个”poison ack”表示这个消息有毒，告诉broker不要再发了，这个时候broker会把这个消息放到DLQ（死信队列）</p>
<p>属性说明</p>
<p><img src="https://gitee.com/RMByuan/note-material/raw/master/studyImg/202112251542731.png" alt="image-20211225154220912"></p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line">手动修改重发次数</span><br><span class="line"></span><br><span class="line">ActiveMQConnection connection ...  <span class="comment">// Create a connection</span></span><br><span class="line"></span><br><span class="line">RedeliveryPolicy queuePolicy = <span class="keyword">new</span> RedeliveryPolicy();</span><br><span class="line">queuePolicy.setInitialRedeliveryDelay(<span class="number">0</span>);  <span class="comment">//初始重发时间间隔</span></span><br><span class="line">queuePolicy.setRedeliveryDelay(<span class="number">1000</span>);  <span class="comment">//重发时间间隔</span></span><br><span class="line">queuePolicy.setUseExponentialBackOff(<span class="keyword">false</span>); <span class="comment">//是否在每次尝试重新发送失败后，增长等待时间</span></span><br><span class="line">queuePolicy.setMaximumRedeliveries(<span class="number">2</span>);  <span class="comment">//重发次数2次</span></span><br><span class="line"></span><br><span class="line">RedeliveryPolicy topicPolicy = <span class="keyword">new</span> RedeliveryPolicy();</span><br><span class="line">topicPolicy.setInitialRedeliveryDelay(<span class="number">0</span>);</span><br><span class="line">topicPolicy.setRedeliveryDelay(<span class="number">1000</span>);</span><br><span class="line">topicPolicy.setUseExponentialBackOff(<span class="keyword">false</span>);</span><br><span class="line">topicPolicy.setMaximumRedeliveries(<span class="number">3</span>);</span><br><span class="line"></span><br><span class="line"><span class="comment">// Receive a message with the JMS API</span></span><br><span class="line">RedeliveryPolicyMap map = connection.getRedeliveryPolicyMap();</span><br><span class="line">map.put(<span class="keyword">new</span> ActiveMQTopic(<span class="string">&quot;&gt;&quot;</span>), topicPolicy);</span><br><span class="line">map.put(<span class="keyword">new</span> ActiveMQQueue(<span class="string">&quot;&gt;&quot;</span>), queuePolicy);</span><br></pre></td></tr></table></figure>

<h2 id="5-死信队列"><a href="#5-死信队列" class="headerlink" title="5.死信队列"></a>5.死信队列</h2><p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/message-redelivery-and-dlq-handling">https://activemq.apache.org/message-redelivery-and-dlq-handling</a></p>
<h3 id="是什么-1"><a href="#是什么-1" class="headerlink" title="是什么"></a>是什么</h3><p>当一条消息被重发的次数超过指定的重发次数，将会被ActiveMQ移入“死信队列”，开发人员可以在这个Queue中查看处理出错的消息，进行人工处理</p>
<h3 id="配置介绍"><a href="#配置介绍" class="headerlink" title="配置介绍"></a>配置介绍</h3><blockquote>
<h3 id="SharedDeadLetterStrategy"><a href="#SharedDeadLetterStrategy" class="headerlink" title="SharedDeadLetterStrategy"></a>SharedDeadLetterStrategy</h3></blockquote>
<p>将所有的DeadLetter保存在一个共享的队列中，这是ActiveMQ broker端默认的策略</p>
<p>共享队列默认为“ActiveMQ.DLQ”,可以通过“deadLetterQueue”属性来设定</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">broker</span>&gt;</span></span><br><span class="line">  </span><br><span class="line">  <span class="tag">&lt;<span class="name">destinationPolicy</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">policyMap</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">policyEntries</span>&gt;</span></span><br><span class="line">        <span class="comment">&lt;!-- Set the following policy on all queues using the &#x27;&gt;&#x27; wildcard --&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">queue</span>=<span class="string">&quot;&gt;&quot;</span>&gt;</span></span><br><span class="line">          <span class="tag">&lt;<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line">              <span class="tag">&lt;<span class="name">shareDeadLetterStrategy</span> <span class="attr">deadLetterQueue</span>=<span class="string">&quot;DLQ-QUEUE&quot;</span>/&gt;</span></span><br><span class="line">          <span class="tag">&lt;/<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;/<span class="name">policyEntry</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">policyEntries</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">policyMap</span>&gt;</span></span><br><span class="line">  <span class="tag">&lt;/<span class="name">destinationPolicy</span>&gt;</span></span><br><span class="line">  </span><br><span class="line"><span class="tag">&lt;/<span class="name">broker</span>&gt;</span></span><br><span class="line"></span><br><span class="line"><span class="tag">&lt;<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line">	</span><br><span class="line"><span class="tag">&lt;/<span class="name">deadLetterStrategy</span>&gt;</span></span><br></pre></td></tr></table></figure>



<blockquote>
<h3 id="individualDeadLetterStrategy"><a href="#individualDeadLetterStrategy" class="headerlink" title="individualDeadLetterStrategy"></a>individualDeadLetterStrategy</h3></blockquote>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br></pre></td><td class="code"><pre><span class="line">  <span class="comment">&lt;!--</span></span><br><span class="line"><span class="comment">          Use the prefix &#x27;DLQ.&#x27; for the destination name, and make</span></span><br><span class="line"><span class="comment">          the DLQ a queue rather than a topic</span></span><br><span class="line"><span class="comment">  --&gt;</span></span><br><span class="line"><span class="tag">&lt;<span class="name">individualDeadLetterStrategy</span> <span class="attr">queuePrefix</span>=<span class="string">&quot;DLQ.&quot;</span> <span class="attr">useQueueForQueueMessages</span>=<span class="string">&quot;true&quot;</span>/&gt;</span></span><br></pre></td></tr></table></figure>

<blockquote>
<h3 id="自动删除过期消息"><a href="#自动删除过期消息" class="headerlink" title="自动删除过期消息"></a><strong>自动删除过期消息</strong></h3></blockquote>
<p>有时需要直接删除过期的消息而不需要发送到死信队列中，“processExpired”表示是否将过期消息放入死信队列，默认为true</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">queue</span>=<span class="string">&quot;&gt;&quot;</span>&gt;</span>   &gt; 类似于sql中的*</span><br><span class="line">    <span class="comment">&lt;!-- </span></span><br><span class="line"><span class="comment">           Tell the dead letter strategy not to process expired messages</span></span><br><span class="line"><span class="comment">           so that they will just be discarded instead of being sent to</span></span><br><span class="line"><span class="comment">           the DLQ </span></span><br><span class="line"><span class="comment">         --&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line">        <span class="tag">&lt;<span class="name">sharedDeadLetterStrategy</span> <span class="attr">processExpired</span>=<span class="string">&quot;false&quot;</span> /&gt;</span></span><br><span class="line">    <span class="tag">&lt;/<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">policyEntry</span>&gt;</span></span><br></pre></td></tr></table></figure>

<blockquote>
<h3 id="将非持久消息放入死信队列"><a href="#将非持久消息放入死信队列" class="headerlink" title="将非持久消息放入死信队列"></a>将非持久消息放入死信队列</h3></blockquote>
<p>默认情况下，ActiveMQ <font color='red'>不会</font>将无法传递的<em>非持久</em>消息放在死信队列中。</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">deadLetterStrategy</span>&gt;</span></span><br><span class="line">    <span class="tag">&lt;<span class="name">sharedDeadLetterStrategy</span> <span class="attr">processNonPersistent</span>=<span class="string">&quot;true&quot;</span> /&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">deadLetterStrategy</span>&gt;</span></span><br></pre></td></tr></table></figure>

<h2 id="6-如何保证消息不被重复消费？（幂等性问题，类似重复提交表单）"><a href="#6-如何保证消息不被重复消费？（幂等性问题，类似重复提交表单）" class="headerlink" title="6.如何保证消息不被重复消费？（幂等性问题，类似重复提交表单）"></a>6.如何保证消息不被重复消费？（幂等性问题，类似重复提交表单）</h2><p>1）如果消息时做数据库的插入操作，给这个消息做一个唯一主键，那么就算出现重复消费的情况，也会导致主键冲突。</p>
<p>2）准备第三方做消费记录，以redis为例，给消息分配一个全局id，只要消费过该消息，将&lt;id,message&gt;以k-v键值对的形式写入redis，那么消费者开始消费前，先去redis中查询有没有消费记录</p>
<h2 id="7-分发策略（Dispatch-Policies）"><a href="#7-分发策略（Dispatch-Policies）" class="headerlink" title="7.分发策略（Dispatch Policies）"></a>7.分发策略（Dispatch Policies）</h2><p>官网：<a target="_blank" rel="noopener" href="https://activemq.apache.org/dispatch-policies">https://activemq.apache.org/dispatch-policies</a></p>
<h3 id="队列的分配策略"><a href="#队列的分配策略" class="headerlink" title="队列的分配策略"></a>队列的分配策略</h3><p>可插入分配策略仅适用于主题。</p>
<p>对于队列，您可以定义分派是否以循环方式发生（默认行为），或者在分派过程选择下一个消费者之前是否将耗尽一个消费者的预取缓冲区（strictOrderDispatch）。</p>
<p>后一种行为是通过<code>strictOrderDispatch</code>在<code>&lt;policyEntry /&gt;</code>元素上设置属性来启用的。例如：</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">queue</span>=<span class="string">&quot;&gt;&quot;</span> <span class="attr">strictOrderDispatch</span>=<span class="string">&quot;false&quot;</span> /&gt;</span></span><br></pre></td></tr></table></figure>

<h3 id="主题的分配策略"><a href="#主题的分配策略" class="headerlink" title="主题的分配策略"></a>主题的分配策略</h3><p>由于调度策略是可插入的，因此主题有更多选项。</p>
<figure class="highlight xml"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br></pre></td><td class="code"><pre><span class="line"><span class="tag">&lt;<span class="name">destinationPolicy</span>&gt;</span></span><br><span class="line">   <span class="tag">&lt;<span class="name">policyMap</span>&gt;</span></span><br><span class="line">      <span class="tag">&lt;<span class="name">policyEntries</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">topic</span>=<span class="string">&quot;FOO.&gt;&quot;</span>&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">dispatchPolicy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">roundRobinDispatchPolicy</span> /&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">dispatchPolicy</span>&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">lastImageSubscriptionRecoveryPolicy</span> /&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;/<span class="name">policyEntry</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">topic</span>=<span class="string">&quot;ORDERS.&gt;&quot;</span>&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">dispatchPolicy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">strictOrderDispatchPolicy</span> /&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">dispatchPolicy</span>&gt;</span></span><br><span class="line">            <span class="comment">&lt;!-- 1 minutes worth --&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">timedSubscriptionRecoveryPolicy</span> <span class="attr">recoverDuration</span>=<span class="string">&quot;60000&quot;</span> /&gt;</span> </span><br><span class="line">            <span class="tag">&lt;/<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;/<span class="name">policyEntry</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">topic</span>=<span class="string">&quot;PRICES.&gt;&quot;</span>&gt;</span></span><br><span class="line">            <span class="comment">&lt;!-- lets force old messages to be discarded for slow consumers --&gt;</span> </span><br><span class="line">            <span class="tag">&lt;<span class="name">pendingMessageLimitStrategy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">constantPendingMessageLimitStrategy</span> <span class="attr">limit</span>=<span class="string">&quot;10&quot;</span>/&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">pendingMessageLimitStrategy</span>&gt;</span></span><br><span class="line">            <span class="comment">&lt;!-- 10 seconds worth --&gt;</span></span><br><span class="line">            <span class="tag">&lt;<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">               <span class="tag">&lt;<span class="name">timedSubscriptionRecoveryPolicy</span> <span class="attr">recoverDuration</span>=<span class="string">&quot;10000&quot;</span> /&gt;</span></span><br><span class="line">            <span class="tag">&lt;/<span class="name">subscriptionRecoveryPolicy</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;/<span class="name">policyEntry</span>&gt;</span></span><br><span class="line">         <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">tempTopic</span>=<span class="string">&quot;true&quot;</span> <span class="attr">advisoryForConsumed</span>=<span class="string">&quot;true&quot;</span> /&gt;</span></span><br><span class="line">         <span class="tag">&lt;<span class="name">policyEntry</span> <span class="attr">tempQueue</span>=<span class="string">&quot;true&quot;</span> <span class="attr">advisoryForConsumed</span>=<span class="string">&quot;true&quot;</span> /&gt;</span></span><br><span class="line">      <span class="tag">&lt;/<span class="name">policyEntries</span>&gt;</span></span><br><span class="line">   <span class="tag">&lt;/<span class="name">policyMap</span>&gt;</span></span><br><span class="line"><span class="tag">&lt;/<span class="name">destinationPolicy</span>&gt;</span></span><br></pre></td></tr></table></figure>

 
      <!-- reward -->
      
      <div id="reword-out">
        <div id="reward-btn">
          打赏
        </div>
      </div>
      
    </div>
    

    <!-- copyright -->
    
    <div class="declare">
      <ul class="post-copyright">
        <li>
          <i class="ri-copyright-line"></i>
          <strong>版权声明： </strong>
          
          本博客所有文章除特别声明外，著作权归作者所有。转载请注明出处！
          
        </li>
      </ul>
    </div>
    
    <footer class="article-footer">
       
<div class="share-btn">
      <span class="share-sns share-outer">
        <i class="ri-share-forward-line"></i>
        分享
      </span>
      <div class="share-wrap">
        <i class="arrow"></i>
        <div class="share-icons">
          
          <a class="weibo share-sns" href="javascript:;" data-type="weibo">
            <i class="ri-weibo-fill"></i>
          </a>
          <a class="weixin share-sns wxFab" href="javascript:;" data-type="weixin">
            <i class="ri-wechat-fill"></i>
          </a>
          <a class="qq share-sns" href="javascript:;" data-type="qq">
            <i class="ri-qq-fill"></i>
          </a>
          <a class="douban share-sns" href="javascript:;" data-type="douban">
            <i class="ri-douban-line"></i>
          </a>
          <!-- <a class="qzone share-sns" href="javascript:;" data-type="qzone">
            <i class="icon icon-qzone"></i>
          </a> -->
          
          <a class="facebook share-sns" href="javascript:;" data-type="facebook">
            <i class="ri-facebook-circle-fill"></i>
          </a>
          <a class="twitter share-sns" href="javascript:;" data-type="twitter">
            <i class="ri-twitter-fill"></i>
          </a>
          <a class="google share-sns" href="javascript:;" data-type="google">
            <i class="ri-google-fill"></i>
          </a>
        </div>
      </div>
</div>

<div class="wx-share-modal">
    <a class="modal-close" href="javascript:;"><i class="ri-close-circle-line"></i></a>
    <p>扫一扫，分享到微信</p>
    <div class="wx-qrcode">
      <img src="//api.qrserver.com/v1/create-qr-code/?size=150x150&data=http://example.com/2021/12/16/ActiveMQ%E7%AC%94%E8%AE%B0/" alt="微信分享二维码">
    </div>
</div>

<div id="share-mask"></div>  
  <ul class="article-tag-list" itemprop="keywords"><li class="article-tag-list-item"><a class="article-tag-list-link" href="/tags/%E6%B6%88%E6%81%AF%E4%B8%AD%E9%97%B4%E4%BB%B6/" rel="tag">消息中间件</a></li></ul>

    </footer>
  </div>

   
  <nav class="article-nav">
    
      <a href="/2022/02/14/SpringMVC%E7%AC%94%E8%AE%B0/" class="article-nav-link">
        <strong class="article-nav-caption">上一篇</strong>
        <div class="article-nav-title">
          
            SpringMVC笔记
          
        </div>
      </a>
    
    
      <a href="/2021/12/11/Spring%E6%B3%A8%E8%A7%A3%E5%BC%80%E5%8F%91/" class="article-nav-link">
        <strong class="article-nav-caption">下一篇</strong>
        <div class="article-nav-title">Spring注解开发</div>
      </a>
    
  </nav>

   
<!-- valine评论 -->
<div id="vcomments-box">
  <div id="vcomments"></div>
</div>
<script src="//cdn1.lncld.net/static/js/3.0.4/av-min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/valine@1.4.14/dist/Valine.min.js"></script>
<script>
  new Valine({
    el: "#vcomments",
    app_id: "",
    app_key: "",
    path: window.location.pathname,
    avatar: "monsterid",
    placeholder: "给我的文章加点评论吧~",
    recordIP: true,
  });
  const infoEle = document.querySelector("#vcomments .info");
  if (infoEle && infoEle.childNodes && infoEle.childNodes.length > 0) {
    infoEle.childNodes.forEach(function (item) {
      item.parentNode.removeChild(item);
    });
  }
</script>
<style>
  #vcomments-box {
    padding: 5px 30px;
  }

  @media screen and (max-width: 800px) {
    #vcomments-box {
      padding: 5px 0px;
    }
  }

  #vcomments-box #vcomments {
    background-color: #fff;
  }

  .v .vlist .vcard .vh {
    padding-right: 20px;
  }

  .v .vlist .vcard {
    padding-left: 10px;
  }
</style>

 
   
     
</article>

</section>
      <footer class="footer">
  <div class="outer">
    <ul>
      <li>
        Copyrights &copy;
        2015-2022
        <i class="ri-heart-fill heart_icon"></i> Yuan
      </li>
    </ul>
    <ul>
      <li>
        
      </li>
    </ul>
    <ul>
      <li>
        
        
        <span>
  <span><i class="ri-user-3-fill"></i>访问人数:<span id="busuanzi_value_site_uv"></span></span>
  <span class="division">|</span>
  <span><i class="ri-eye-fill"></i>浏览次数:<span id="busuanzi_value_page_pv"></span></span>
</span>
        
      </li>
    </ul>
    <ul>
      
    </ul>
    <ul>
      
    </ul>
    <ul>
      <li>
        <!-- cnzz统计 -->
        
        <script type="text/javascript" src='https://s9.cnzz.com/z_stat.php?id=1278069914&amp;web_id=1278069914'></script>
        
      </li>
    </ul>
  </div>
</footer>    
    </main>
    <div class="float_btns">
      <div class="totop" id="totop">
  <i class="ri-arrow-up-line"></i>
</div>

<div class="todark" id="todark">
  <i class="ri-moon-line"></i>
</div>

    </div>
    <aside class="sidebar on">
      <button class="navbar-toggle"></button>
<nav class="navbar">
  
  <div class="logo">
    <a href="/"><img src="/images/ayer-side.svg" alt="MyBlog"></a>
  </div>
  
  <ul class="nav nav-main">
    
    <li class="nav-item">
      <a class="nav-item-link" href="/">主页</a>
    </li>
    
    <li class="nav-item">
      <a class="nav-item-link" href="/archives">归档</a>
    </li>
    
    <li class="nav-item">
      <a class="nav-item-link" href="/categories">分类</a>
    </li>
    
    <li class="nav-item">
      <a class="nav-item-link" href="/tags">标签</a>
    </li>
    
    <li class="nav-item">
      <a class="nav-item-link" href="/friends">友链</a>
    </li>
    
    <li class="nav-item">
      <a class="nav-item-link" href="/about">关于我</a>
    </li>
    
  </ul>
</nav>
<nav class="navbar navbar-bottom">
  <ul class="nav">
    <li class="nav-item">
      
      <a class="nav-item-link nav-item-search"  title="搜索">
        <i class="ri-search-line"></i>
      </a>
      
      
      <a class="nav-item-link" target="_blank" href="/atom.xml" title="RSS Feed">
        <i class="ri-rss-line"></i>
      </a>
      
    </li>
  </ul>
</nav>
<div class="search-form-wrap">
  <div class="local-search local-search-plugin">
  <input type="search" id="local-search-input" class="local-search-input" placeholder="Search...">
  <div id="local-search-result" class="local-search-result"></div>
</div>
</div>
    </aside>
    <div id="mask"></div>

<!-- #reward -->
<div id="reward">
  <span class="close"><i class="ri-close-line"></i></span>
  <p class="reward-p"><i class="ri-cup-line"></i>请我喝杯咖啡吧~</p>
  <div class="reward-box">
    
    
  </div>
</div>
    
<script src="/js/jquery-3.6.0.min.js"></script>
 
<script src="/js/lazyload.min.js"></script>

<!-- Tocbot -->
 
<script src="/js/tocbot.min.js"></script>

<script>
  tocbot.init({
    tocSelector: ".tocbot",
    contentSelector: ".article-entry",
    headingSelector: "h1, h2, h3, h4, h5, h6",
    hasInnerContainers: true,
    scrollSmooth: true,
    scrollContainer: "main",
    positionFixedSelector: ".tocbot",
    positionFixedClass: "is-position-fixed",
    fixedSidebarOffset: "auto",
  });
</script>

<script src="https://cdn.jsdelivr.net/npm/jquery-modal@0.9.2/jquery.modal.min.js"></script>
<link
  rel="stylesheet"
  href="https://cdn.jsdelivr.net/npm/jquery-modal@0.9.2/jquery.modal.min.css"
/>
<script src="https://cdn.jsdelivr.net/npm/justifiedGallery@3.7.0/dist/js/jquery.justifiedGallery.min.js"></script>

<script src="/dist/main.js"></script>

<!-- ImageViewer -->
 <!-- Root element of PhotoSwipe. Must have class pswp. -->
<div class="pswp" tabindex="-1" role="dialog" aria-hidden="true">

    <!-- Background of PhotoSwipe. 
         It's a separate element as animating opacity is faster than rgba(). -->
    <div class="pswp__bg"></div>

    <!-- Slides wrapper with overflow:hidden. -->
    <div class="pswp__scroll-wrap">

        <!-- Container that holds slides. 
            PhotoSwipe keeps only 3 of them in the DOM to save memory.
            Don't modify these 3 pswp__item elements, data is added later on. -->
        <div class="pswp__container">
            <div class="pswp__item"></div>
            <div class="pswp__item"></div>
            <div class="pswp__item"></div>
        </div>

        <!-- Default (PhotoSwipeUI_Default) interface on top of sliding area. Can be changed. -->
        <div class="pswp__ui pswp__ui--hidden">

            <div class="pswp__top-bar">

                <!--  Controls are self-explanatory. Order can be changed. -->

                <div class="pswp__counter"></div>

                <button class="pswp__button pswp__button--close" title="Close (Esc)"></button>

                <button class="pswp__button pswp__button--share" style="display:none" title="Share"></button>

                <button class="pswp__button pswp__button--fs" title="Toggle fullscreen"></button>

                <button class="pswp__button pswp__button--zoom" title="Zoom in/out"></button>

                <!-- Preloader demo http://codepen.io/dimsemenov/pen/yyBWoR -->
                <!-- element will get class pswp__preloader--active when preloader is running -->
                <div class="pswp__preloader">
                    <div class="pswp__preloader__icn">
                        <div class="pswp__preloader__cut">
                            <div class="pswp__preloader__donut"></div>
                        </div>
                    </div>
                </div>
            </div>

            <div class="pswp__share-modal pswp__share-modal--hidden pswp__single-tap">
                <div class="pswp__share-tooltip"></div>
            </div>

            <button class="pswp__button pswp__button--arrow--left" title="Previous (arrow left)">
            </button>

            <button class="pswp__button pswp__button--arrow--right" title="Next (arrow right)">
            </button>

            <div class="pswp__caption">
                <div class="pswp__caption__center"></div>
            </div>

        </div>

    </div>

</div>

<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/photoswipe@4.1.3/dist/photoswipe.min.css">
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/photoswipe@4.1.3/dist/default-skin/default-skin.min.css">
<script src="https://cdn.jsdelivr.net/npm/photoswipe@4.1.3/dist/photoswipe.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/photoswipe@4.1.3/dist/photoswipe-ui-default.min.js"></script>

<script>
    function viewer_init() {
        let pswpElement = document.querySelectorAll('.pswp')[0];
        let $imgArr = document.querySelectorAll(('.article-entry img:not(.reward-img)'))

        $imgArr.forEach(($em, i) => {
            $em.onclick = () => {
                // slider展开状态
                // todo: 这样不好，后面改成状态
                if (document.querySelector('.left-col.show')) return
                let items = []
                $imgArr.forEach(($em2, i2) => {
                    let img = $em2.getAttribute('data-idx', i2)
                    let src = $em2.getAttribute('data-target') || $em2.getAttribute('src')
                    let title = $em2.getAttribute('alt')
                    // 获得原图尺寸
                    const image = new Image()
                    image.src = src
                    items.push({
                        src: src,
                        w: image.width || $em2.width,
                        h: image.height || $em2.height,
                        title: title
                    })
                })
                var gallery = new PhotoSwipe(pswpElement, PhotoSwipeUI_Default, items, {
                    index: parseInt(i)
                });
                gallery.init()
            }
        })
    }
    viewer_init()
</script> 
<!-- MathJax -->

<!-- Katex -->

<!-- busuanzi  -->
 
<script src="/js/busuanzi-2.3.pure.min.js"></script>
 
<!-- ClickLove -->
 
<script src="/js/clickLove.js"></script>
 
<!-- ClickBoom1 -->

<!-- ClickBoom2 -->

<!-- CodeCopy -->
 
<link rel="stylesheet" href="/css/clipboard.css">
 <script src="https://cdn.jsdelivr.net/npm/clipboard@2/dist/clipboard.min.js"></script>
<script>
  function wait(callback, seconds) {
    var timelag = null;
    timelag = window.setTimeout(callback, seconds);
  }
  !function (e, t, a) {
    var initCopyCode = function(){
      var copyHtml = '';
      copyHtml += '<button class="btn-copy" data-clipboard-snippet="">';
      copyHtml += '<i class="ri-file-copy-2-line"></i><span>COPY</span>';
      copyHtml += '</button>';
      $(".highlight .code pre").before(copyHtml);
      $(".article pre code").before(copyHtml);
      var clipboard = new ClipboardJS('.btn-copy', {
        target: function(trigger) {
          return trigger.nextElementSibling;
        }
      });
      clipboard.on('success', function(e) {
        let $btn = $(e.trigger);
        $btn.addClass('copied');
        let $icon = $($btn.find('i'));
        $icon.removeClass('ri-file-copy-2-line');
        $icon.addClass('ri-checkbox-circle-line');
        let $span = $($btn.find('span'));
        $span[0].innerText = 'COPIED';
        
        wait(function () { // 等待两秒钟后恢复
          $icon.removeClass('ri-checkbox-circle-line');
          $icon.addClass('ri-file-copy-2-line');
          $span[0].innerText = 'COPY';
        }, 2000);
      });
      clipboard.on('error', function(e) {
        e.clearSelection();
        let $btn = $(e.trigger);
        $btn.addClass('copy-failed');
        let $icon = $($btn.find('i'));
        $icon.removeClass('ri-file-copy-2-line');
        $icon.addClass('ri-time-line');
        let $span = $($btn.find('span'));
        $span[0].innerText = 'COPY FAILED';
        
        wait(function () { // 等待两秒钟后恢复
          $icon.removeClass('ri-time-line');
          $icon.addClass('ri-file-copy-2-line');
          $span[0].innerText = 'COPY';
        }, 2000);
      });
    }
    initCopyCode();
  }(window, document);
</script>
 
<!-- CanvasBackground -->

<script>
  if (window.mermaid) {
    mermaid.initialize({ theme: "forest" });
  }
</script>


    
    <div id="music">
    
    
    
    <iframe frameborder="no" border="1" marginwidth="0" marginheight="0" width="200" height="52"
        src="//music.163.com/outchain/player?type=2&id=22707008&auto=1&height=32"></iframe>
</div>

<style>
    #music {
        position: fixed;
        right: 15px;
        bottom: 0;
        z-index: 998;
    }
</style>
    
    

  </div>
</body>

</html>